distsys hw2

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)
}