NativeIOCPServer_V1.h

Go to the documentation of this file.

// Copyright 2024 libsese
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//    http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
 
#pragma once
 
#include <sese/net/Address.h>
#include <sese/security/SSLContext.h>
#include <sese/service/iocp/IOBuf.h>
#include <sese/thread/Thread.h>
#include <sese/io/ByteBuilder.h>
#include <sese/util/TimeWheel.h>
 
#include <atomic>
#include <set>
#include <mutex>
 
namespace sese::_windows::iocp::v1 {
 
class NativeIOCPServer;
struct OverlappedWrapper;
 
class NativeContext final : public io::InputStream, public io::OutputStream, public io::PeekableStream {
    friend class NativeIOCPServer;
    using IOBuf = sese::iocp::IOBuf;
    using Node = std::unique_ptr<sese::iocp::IOBufNode>;
 
    enum class Type {
        READ,
        WRITE,
        CONNECT,
        READY,
        CLOSE
    };
 
    OverlappedWrapper *pWrapper{};
    WSABUF wsabufRead{};
    WSABUF wsabufWrite{};
    Type type{Type::READ};
    SOCKET fd{INVALID_SOCKET};
    NativeIOCPServer *self{};
    TimeoutEvent *timeoutEvent{};
    void *ssl{};
    void *bio{};
    Node readNode;
    IOBuf recv{};
    io::ByteBuilder send{IOCP_WSABUF_SIZE, IOCP_WSABUF_SIZE};
    void *data{};
 
public:
    explicit NativeContext(OverlappedWrapper *p_wrapper);
    ~NativeContext() override;
    int64_t read(void *buffer, size_t length) override;
    int64_t write(const void *buffer, size_t length) override;
    int64_t peek(void *buffer, size_t length) override;
    int64_t trunc(size_t length) override;
    [[nodiscard]] int32_t getFd() const { return static_cast<int32_t>(NativeContext::fd); }
    [[nodiscard]] void *getData() const { return NativeContext::data; }
    void setData(void *p_data) { NativeContext::data = p_data; }
};
 
struct OverlappedWrapper final {
    OVERLAPPED overlapped{};
    NativeContext ctx;
 
    OverlappedWrapper();
};
 
class NativeIOCPServer {
public:
    using Context = NativeContext;
    using DeleteContextCallback = std::function<void(Context *data)>;
 
    virtual ~NativeIOCPServer() = default;
 
    bool init();
    void shutdown();
    void postRead(Context *ctx);
    void postWrite(Context *ctx);
    void postClose(Context *ctx);
    void postConnect(const net::IPAddress::Ptr &to, const security::SSLContext::Ptr &cli_ctx, void *data = nullptr);
    void setTimeout(Context *ctx, int64_t seconds);
    void cancelTimeout(Context *ctx);
    static void onDeleteContext(Context *) {}
    virtual void onAcceptCompleted(Context *ctx) {}
    virtual void onPreRead(Context *ctx) {}
    virtual void onReadCompleted(Context *ctx) {}
    virtual void onWriteCompleted(Context *ctx) {}
    virtual void onTimeout(Context *ctx) {}
    virtual void onPreConnect(Context *ctx) {}
    virtual void onConnected(Context *ctx){};
    virtual void onAlpnGet(Context *ctx, const uint8_t *in, uint32_t in_length){};
    int onAlpnSelect(
            const uint8_t **out, uint8_t *out_length,
            const uint8_t *in, uint32_t in_length
    );
 
public:
    void setAddress(const net::IPAddress::Ptr &addr) { NativeIOCPServer::address = addr; }
    void setThreads(size_t number_of_threads) { NativeIOCPServer::threads = number_of_threads; }
    void setServCtx(const security::SSLContext::Ptr &ctx) { NativeIOCPServer::sslCtx = ctx; }
    void setServProtos(const std::string &protos) { NativeIOCPServer::servProtos = protos; }
    void setClientProtos(const std::string &protos) { NativeIOCPServer::clientProtos = protos; }
    void setDeleteContextCallback(const DeleteContextCallback &callback) { NativeIOCPServer::deleteContextCallback = callback; }
    [[nodiscard]] const security::SSLContext::Ptr &getServCtx() const { return NativeIOCPServer::sslCtx; }
    [[nodiscard]] const DeleteContextCallback &getDeleteContextCallback() const { return NativeIOCPServer::deleteContextCallback; };
    bool isActiveReleaseMode() const { return NativeIOCPServer::activeReleaseMode; }
 
protected:
    void setActiveReleaseMode(bool enable) { NativeIOCPServer::activeReleaseMode = enable; }
    void releaseContext(Context *ctx);
 
    void acceptThreadProc();
    void eventThreadProc();
    static int alpnCallbackFunction(
            void *ssl,
            const uint8_t **out, uint8_t *out_length,
            const uint8_t *in, uint32_t in_length,
            NativeIOCPServer *server
    );
 
    void *connectEx{};
    bool initConnectEx();
 
    static long bioCtrl(void *bio, int cmd, long num, void *ptr);
    static int bioWrite(void *bio, const char *in, int length);
    static int bioRead(void *bio, char *out, int length);
 
    std::atomic_bool isShutdown{false};
    HANDLE iocpFd{INVALID_HANDLE_VALUE};
    SOCKET listenFd{INVALID_SOCKET};
    net::IPAddress::Ptr address{};
    Thread::Ptr acceptThread{};
 
    TimeWheel wheel{};
    std::set<OverlappedWrapper *> wrapperSet{};
    std::mutex wrapperSetMutex{};
 
    size_t threads{2};
    std::vector<Thread::Ptr> eventThreadGroup{};
    DeleteContextCallback deleteContextCallback = onDeleteContext;
    security::SSLContext::Ptr sslCtx{};
    void *bioMethod{};
    std::string servProtos{};
    std::string clientProtos{};
 
private:
    bool activeReleaseMode = true;
};
 
} // namespace sese::_windows::iocp::v1