Go中来自客户端和服务器的RPC

Question

Is it actually possible to do RPC calls from a server to a client with the net/rpc package in Go? If no, is there a better solution out there?

Answer 1

I am currently using thrift (thrift4go) for server->client and client->server RPC functionality. By default, thrift does only client->server calls just like net/rpc. As I also required server->client communication, I did some research and found bidi-thrift. Bidi-thrift explains how to connect a java server + java client to have bidirectional thrift communication.

What bidi-thrift does, and it's limitations.

A TCP connection has an incomming and outgoing communication line (RC and TX). The idea of bidi-thrift is to split RS and TX and provide these to a server(processor) and client(remote) on both client-application and server-application. I found this to be hard to do in Go. Also, this way there is no "response" possible (the response line is in use). Therefore, all methods in the service's must be "oneway void". (fire and forget, call gives no result).

The solution

I changed the idea of bidi-thrift and made the client open two connections to the server, A and B. The first connection(A) is used to perform client -> server communication (where client makes the calls, as usual). The second connection(B) is 'hijacked', and connected to a server(processor) on the client, while it is connected to a client(remote) on the server. I've got this working with a Go server and a Java client. It works very well. It's fast and reliable (just like normal thrift is).

Some sources.. The B connection (server->client) is set up like this:

Go server

// factories
framedTransportFactory := thrift.NewTFramedTransportFactory(thrift.NewTTransportFactory())
protocolFactory := thrift.NewTBinaryProtocolFactoryDefault()

// create socket listener
addr, err := net.ResolveTCPAddr("tcp", "127.0.0.1:9091")
if err != nil {
    log.Print("Error resolving address: ", err.Error(), "
")
    return
}
serverTransport, err := thrift.NewTServerSocketAddr(addr)
if err != nil {
    log.Print("Error creating server socket: ", err.Error(), "
")
    return
}

// Start the server to listen for connections
log.Print("Starting the server for B communication (server->client) on ", addr, "
")
err = serverTransport.Listen()
if err != nil {
    log.Print("Error during B server: ", err.Error(), "
")
    return //err
}

// Accept new connections and handle those
for {
    transport, err := serverTransport.Accept()
    if err != nil {
        return //err
    }
    if transport != nil {
        // Each transport is handled in a goroutine so the server is availiable again.
        go func() {
            useTransport := framedTransportFactory.GetTransport(transport)
            client := worldclient.NewWorldClientClientFactory(useTransport, protocolFactory)

            // Thats it!
            // Lets do something with the connction
            result, err := client.Hello()
            if err != nil {
                log.Printf("Errror when calling Hello on client: %s
", err)
            }

            // client.CallSomething()
        }()
    }
}

Java client

// preparations for B connection
TTransportFactory transportFactory = new TTransportFactory();
TProtocolFactory protocolFactory = new TBinaryProtocol.Factory();
YourServiceProcessor processor = new YourService.Processor<YourServiceProcessor>(new YourServiceProcessor(this));


/* Create thrift connection for B calls (server -> client) */
try {
    // create the transport
    final TTransport transport = new TSocket("127.0.0.1", 9091);

    // open the transport
    transport.open();

    // add framing to the transport layer
    final TTransport framedTransport = new TFramedTransport(transportFactory.getTransport(transport));

    // connect framed transports to protocols
    final TProtocol protocol = protocolFactory.getProtocol(framedTransport);

    // let the processor handle the requests in new Thread
    new Thread() {
        public void run() {
            try {
                while (processor.process(protocol, protocol)) {}
            } catch (TException e) {
                e.printStackTrace();
            } catch (NullPointerException e) {
                e.printStackTrace();
            }
        }
    }.start();
} catch(Exception e) {
    e.printStackTrace();
}