双向通讯

例程：bidirectional

正常情况下，客户端发送请求给服务，服务返回响应给客户端。这是一个request-responserpc模型。

但是对于一些用户，他们希望服务可以主动发送指令或者通知给客户端。这可以通过在之前的客户端安装一个服务、并且在之前的服务端安装一个客户端来实现。但是这样做既冗余又复杂。

rpcx实现了一个简单的通知模型。

你应该缓存连接和事务用户ID来确保你知道你想发送通知给哪个客户端。

服务器

服务器可以使用SendMessage来发送消息给客户端，数据类型为[]byte。你应该使用servicePath和serviceMethod来指明数据发送给哪一个通知。

你可以在服务中使用ctx.Value(server.RemoteConnContextKey)获取到net.Conn。

func (s *Server) SendMessage(conn net.Conn, servicePath, serviceMethod string, metadata map[string]string, data []byte) error

// go server.go 

func main() {
    flag.Parse()
    ln, _ := net.Listen("tcp", ":9981")
    go http.Serve(ln, nil)

    s := server.NewServer()
    //s.RegisterName("Arith", new(example.Arith), "")
    s.Register(new(Arith), "")
    go s.Serve("tcp", *addr)

    for !connected {
        time.Sleep(time.Second)
    }

    fmt.Printf("start to send messages to %s\n", clientConn.RemoteAddr().String())
    for {
        if clientConn != nil {
            err := s.SendMessage(clientConn, "test_service_path", "test_service_method", nil, []byte("abcde"))
            if err != nil {
                fmt.Printf("failed to send messsage to %s: %v\n", clientConn.RemoteAddr().String(), err)
                clientConn = nil
            }
        }
        time.Sleep(time.Second)
    }
}

客户端

你必须使用NewBidirectionalXClient传递一个通道来创建一个XClient。然后你就可以遍历通道来获取消息。

// go client.go

func main() {
    flag.Parse()

    ch := make(chan *protocol.Message)

    d := client.NewPeer2PeerDiscovery("tcp@"+*addr, "")
    xclient := client.NewBidirectionalXClient("Arith", client.Failtry, client.RandomSelect, d, client.DefaultOption, ch)
    defer xclient.Close()

    args := &example.Args{
        A: 10,
        B: 20,
    }

    reply := &example.Reply{}
    err := xclient.Call(context.Background(), "Mul", args, reply)
    if err != nil {
        log.Fatalf("failed to call: %v", err)
    }

    log.Printf("%d * %d = %d", args.A, args.B, reply.C)

    for msg := range ch {
        fmt.Printf("receive msg from server: %s\n", msg.Payload)
    }
}