package main import ( "context" "fmt" "log" "net" "os" "sync" pb "dagorokhov/messenger/proto" emptypb "google.golang.org/protobuf/types/known/emptypb" timestamppb "google.golang.org/protobuf/types/known/timestamppb" grpc "google.golang.org/grpc" ) type messengerServer struct { pb.UnimplementedMessengerServerServer channels []chan *pb.Message mutex sync.Mutex } func (s *messengerServer) SendMessage(ctx context.Context, message *pb.Message) (*pb.Reply, error) { fmt.Println("got message", message) reply := pb.Reply{SendTime: timestamppb.Now()} message.SendTime = reply.SendTime s.mutex.Lock() defer s.mutex.Unlock() for _, channel := range s.channels { channel <- message } return &reply, nil } func (s *messengerServer) ReadMessages(_ *emptypb.Empty, stream grpc.ServerStreamingServer[pb.Message]) error { fmt.Println("connected to listen") s.mutex.Lock() s.channels = append(s.channels, make(chan *pb.Message)) channel := s.channels[len(s.channels)-1] s.mutex.Unlock() for message := range channel { fmt.Println("trying to send message", message) if err := stream.Send(message); err != nil { fmt.Println(err) return err } } return nil } func main() { fmt.Println("starting server") port := os.Getenv("MESSENGER_SERVER_PORT") if port == "" { port = "51075" } lis, err := net.Listen("tcp", fmt.Sprintf("localhost:%v", port)) if err != nil { log.Fatalf("failed to listen: %v", err) } var opts []grpc.ServerOption grpcServer := grpc.NewServer(opts...) server := messengerServer{} pb.RegisterMessengerServerServer(grpcServer, &server) grpcServer.Serve(lis) }