POCO - 我可以使用 SocketReactor 作为客户端吗？

Question

当人们运行 TCP 服务器时，我看到很多将 SocketReactor 与 SocketAcceptor 结合使用的示例。

但是，我希望作为客户端连接到现有的 TCP 服务器，但希望能够处理 SocketReactor 公开的事件，例如 onSocketReadable、onSocketWritable、onSocketShutdown、onSocketError 和 onSocketTimeout。

这可以通过 POCO 库实现吗？我将以下代码放在一起，但没有触发任何事件。 如果这种方法不起作用，还有其他建议吗？ 基本上我会从服务器接收实时的 tcp 消息流，我也会将消息发送回服务器以执行某些任务。

class ITCHProvider
    {
    private:
        Poco::Net::SocketAddress _sa;
        Poco::Net::StreamSocket _sock;
        Poco::Net::SocketStream _stream;
        Poco::Net::SocketReactor _reactor;

    public:
        ITCHProvider() :
            _sa("host", 1234),
            _sock(),
            _stream(_sock),
            _reactor()
        {
            _reactor.addEventHandler(_sock, Poco::NObserver<ITCHProvider, Poco::Net::ReadableNotification>(*this, &ITCHProvider::onSocketReadable));
            _reactor.addEventHandler(_sock, Poco::NObserver<ITCHProvider, Poco::Net::WritableNotification>(*this, &ITCHProvider::onSocketWritable));
            _reactor.addEventHandler(_sock, Poco::NObserver<ITCHProvider, Poco::Net::ShutdownNotification>(*this, &ITCHProvider::onSocketShutdown));
            _reactor.addEventHandler(_sock, Poco::NObserver<ITCHProvider, Poco::Net::ErrorNotification>(*this, &ITCHProvider::onSocketError));
            _reactor.addEventHandler(_sock, Poco::NObserver<ITCHProvider, Poco::Net::TimeoutNotification>(*this, &ITCHProvider::onSocketTimeout));

            _sock.connect(_sa);
        }

        ~ITCHProvider()
        {
            close();
        }

        void onSocketReadable(const Poco::AutoPtr<Poco::Net::ReadableNotification>& pNf)
        {
            LOG(INFO) << "READable   !!";
        }
        void onSocketWritable(const Poco::AutoPtr<Poco::Net::WritableNotification>& pNf)
        {
            LOG(INFO) << "WRITEable   !!";
        }
        void onSocketShutdown(const Poco::AutoPtr<Poco::Net::ShutdownNotification>& pNf)
        {
            LOG(INFO) << "SHUTDOWN!!!!!!!!!!!!";
        }
        void onSocketError(const Poco::AutoPtr<Poco::Net::ErrorNotification>& pNf)
        {
            LOG(INFO) << "Error!!";
        }
        void onSocketTimeout(const Poco::AutoPtr<Poco::Net::TimeoutNotification>& pNf)
        {
            LOG(INFO) << "Timeout!!";
        }

        // Close down the connection properly.
        void close() {
            try {
                _sock.shutdown();
            }
            catch (...) {
                LOG(INFO) << "closing failed.";
            }
        }

    };

Answer 1

可以做到，但请注意 (a) 在您实际运行反应器之前不会发生任何事情，并且 (b) 反应器将旋转直到明确停止，因此最好在单独的线程中运行它；

这是一个简单的例子：

// server-side handler
class EchoServiceHandler {
public:
    EchoServiceHandler(StreamSocket& socket, SocketReactor& reactor): _socket(socket), _reactor(reactor) {
        _reactor.addEventHandler(_socket, Observer<EchoServiceHandler, ReadableNotification>(*this, &EchoServiceHandler::onReadable));
    }

    ~EchoServiceHandler() {
        _reactor.removeEventHandler(_socket, Observer<EchoServiceHandler, ReadableNotification>(*this, &EchoServiceHandler::onReadable));
    }

    void onReadable(ReadableNotification* pNf) {
        pNf->release();
        char buffer[8];
        int n = _socket.receiveBytes(buffer, sizeof(buffer));
        if (n > 0) {
            _socket.sendBytes(buffer, n);
        }
        else {
            _socket.shutdownSend();
            delete this;
        }
    }

    private:
        StreamSocket   _socket;
        SocketReactor& _reactor;
};

您的原始代码，稍作修改：

class ITCHProvider : public Poco::Runnable
{
private:
    Poco::Net::StreamSocket _sock;
    Poco::Net::SocketReactor _reactor;

public:
    ITCHProvider(const SocketAddress& sa) : _sock(sa) {
        _reactor.addEventHandler(_sock, Poco::NObserver<ITCHProvider, Poco::Net::ReadableNotification>(*this, &ITCHProvider::onSocketReadable));
        _reactor.addEventHandler(_sock, Poco::NObserver<ITCHProvider, Poco::Net::WritableNotification>(*this, &ITCHProvider::onSocketWritable));
        _reactor.addEventHandler(_sock, Poco::NObserver<ITCHProvider, Poco::Net::ShutdownNotification>(*this, &ITCHProvider::onSocketShutdown));
        _reactor.addEventHandler(_sock, Poco::NObserver<ITCHProvider, Poco::Net::ErrorNotification>(*this, &ITCHProvider::onSocketError));
        _reactor.addEventHandler(_sock, Poco::NObserver<ITCHProvider, Poco::Net::TimeoutNotification>(*this, &ITCHProvider::onSocketTimeout));
        std::string data = "Hello reactor world!";
        _sock.sendBytes(data.data(), (int)data.length());
    }

    ~ITCHProvider() { close(); }

    void run() { _reactor.run(); }
    void stop() { _reactor.stop(); }

    void onSocketReadable(const Poco::AutoPtr<Poco::Net::ReadableNotification>& pNf) {
        std::cout << "READable   !!" << std::endl;
        char data[1025] = { 0 };
        if (_sock.receiveBytes(data, 1024) > 0) {
            std::cout << data << std::endl;
        }
    }
    void onSocketWritable(const Poco::AutoPtr<Poco::Net::WritableNotification>& pNf) {
        std::cout << "WRITEable   !!" << std::endl;
    }
    void onSocketShutdown(const Poco::AutoPtr<Poco::Net::ShutdownNotification>& pNf) {
        std::cout << "SHUTDOWN!!!!!!!!!!!!" << std::endl;
    }
    void onSocketError(const Poco::AutoPtr<Poco::Net::ErrorNotification>& pNf) {
        std::cout << "Error!!" << std::endl;
    }
    void onSocketTimeout(const Poco::AutoPtr<Poco::Net::TimeoutNotification>& pNf) {
        std::cout << "Timeout!!" << std::endl;
    }

    void close() {
        try {
            _sock.shutdown();
        }
        catch (...) {
            std::cout << "closing failed." << std::endl;
        }
    }
};

现在，让我们运行上面的代码：

SocketAddress ssa;
ServerSocket ss(ssa);
SocketReactor reactor;
SocketAcceptor<EchoServiceHandler> acceptor(ss, reactor);
Thread server;
server.start(reactor);
ITCHProvider provider(SocketAddress("127.0.0.1", ss.address().port()));
Thread client;
client.start(provider);
Thread::sleep(1000);
reactor.stop();
provider.stop();
server.join();
client.join();

输出（为了清楚起见，可写和超时通知静音）：

READable   !!
Hello reactor world!
SHUTDOWN!!!!!!!!!!!!