test/grpc-odbserver/matrix/topo.go

package mnode

import (
	"context"
	"fmt"
	"log"
	"net"
	"time"
	"trial/grpc-odbserver/grpc/api"

	"google.golang.org/grpc"
	"gopkg.in/yaml.v3"
)

type MeshServer struct {
	api.UnimplementedMeshServer
}

func (ts *MeshServer) Broadcast(tss api.Mesh_BroadcastServer) error {
	for {
		ti, e := tss.Recv()
		if e != nil {
			return e
		}
		bs, _ := yaml.Marshal(ti.GetNodeInfo())
		fmt.Println(string(bs))
	}
}

func (ts *MeshServer) Unicast(tss api.Mesh_UnicastServer) error {
	for {
		ti, e := tss.Recv()
		if e != nil {
			return e
		}
		bs, _ := yaml.Marshal(ti.GetNodeInfo())
		fmt.Println(string(bs))
	}
}

var listen_ready = make(chan struct{})

func server() {
	svr := grpc.NewServer()
	api.RegisterMeshServer(svr, &MeshServer{})
	lis, _ := net.Listen("tcp", ":11111")
	listen_ready <- struct{}{}
	log.Printf("listen: %v", lis.Addr())
	err := svr.Serve(lis)
	if err != nil {
		log.Printf("did not connect: %v", err)
	}
}

func client() {
	cc, err := grpc.Dial("127.0.0.1:11111", grpc.WithInsecure())
	if err != nil {
		log.Printf("did not connect: %v", err)
	}
	defer cc.Close()
	tc := api.NewMeshClient(cc)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	tsc, _ := tc.Broadcast(ctx)
	n := 0
	for {
		n++
		tsc.Send(&api.MeshData{
			Message: &api.MeshData_NodeInfo{NodeInfo: &api.NodeInfo{
				NodeID:   time.Now().UnixNano(),
				NodeName: fmt.Sprint("xxx", n),
			}},
		})
		log.Print("send", n)
		time.Sleep(5 * time.Second)
	}
}

func main() {
	go server()
	<-listen_ready
	client()
}