Bidirectional commnunication

Example:bidirectional

In normal case, clients send requests to services and services returns reponses to clients. It is the request-response rpc model.

But for some users, they want services can send commands or notifications to clients initiatively. It can be implemented by installing a service on the prior client and a client on the prior service but it is redundant and complicated.

Rpcx implements a simple notification model.

You should cache the connection and business user ID in order that yu know you want to send notifications to which client.

Server

Server can use SendMessage to send messages to clients and data is []byte. You should use servicePath and serviceMethod to indicate which notification the data is.

You can get the net.Conn by ctx.Value(server.RemoteConnContextKey) in service.

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

```go server.gofunc main() { flag.Parse()

  1. ln, _ := net.Listen("tcp", ":9981")
  2. go http.Serve(ln, nil)
  4. s := server.NewServer()
  5. //s.RegisterName("Arith", new(example.Arith), "")
  6. s.Register(new(Arith), "")
  7. go s.Serve("tcp", *addr)
  9. for !connected {
  10.     time.Sleep(time.Second)
  11. }
  13. fmt.Printf("start to send messages to %s\n", clientConn.RemoteAddr().String())
  14. for {
  15.     if clientConn != nil {
  16.         err := s.SendMessage(clientConn, "test_service_path", "test_service_method", nil, []byte("abcde"))
  17.         if err != nil {
  18.             fmt.Printf("failed to send messsage to %s: %v\n", clientConn.RemoteAddr().String(), err)
  19.             clientConn = nil
  20.         }
  21.     }
  22.     time.Sleep(time.Second)
  23. }

}

  2. ## Client
  4. You must use `NewBidirectionalXClient` to create XClient by passing a channel. Then yuou can range the channel to get the message.
  6. ```go client.go
  7. func main() {
  8.     flag.Parse()
  10.     ch := make(chan *protocol.Message)
  12.     d := client.NewPeer2PeerDiscovery("tcp@"+*addr, "")
  13.     xclient := client.NewBidirectionalXClient("Arith", client.Failtry, client.RandomSelect, d, client.DefaultOption, ch)
  14.     defer xclient.Close()
  16.     args := &example.Args{
  17.         A: 10,
  18.         B: 20,
  19.     }
  21.     reply := &example.Reply{}
  22.     err := xclient.Call(context.Background(), "Mul", args, reply)
  23.     if err != nil {
  24.         log.Fatalf("failed to call: %v", err)
  25.     }
  27.     log.Printf("%d * %d = %d", args.A, args.B, reply.C)
  29.     for msg := range ch {
  30.         fmt.Printf("receive msg from server: %s\n", msg.Payload)
  31.     }
  32. }

By smallnest updated 2018-12-04 11:47:26