server.js

const grpc = require("@grpc/grpc-js");
const protoLoader = require("@grpc/proto-loader");
const path = require("path");

// Load the protobuf definition
const PROTO_PATH = path.join(__dirname, "chat.proto");
const packageDefinition = protoLoader.loadSync(PROTO_PATH, {
  keepCase: true,
  longs: String,
  enums: String,
  defaults: true,
  oneofs: true,
});
const chat_proto = grpc.loadPackageDefinition(packageDefinition).chat;

// In-memory array to store all active client calls (streams)
const connectedClients = [];

/**
 * Implements the Chat RPC method.
 * This is a bidirectional streaming RPC.
 * @param {grpc.ServerDuplexStream} call The client's stream
 */
function chat(call) {
  const clientId = Date.now().toString(); // Simple unique ID for the client
  console.log(`Client ${clientId} connected for chat.`);

  // Add the current client's stream to the list of connected clients
  connectedClients.push(call);

  // Handle incoming messages from this client
  call.on("data", (message) => {
    const chatMessage = {
      sender: message.sender || "Unknown",
      message: message.message,
      timestamp: Date.now(),
    };
    console.log(`[${chatMessage.sender}]: ${chatMessage.message}`);

    // Broadcast the received message to all other connected clients
    connectedClients.forEach((clientCall) => {
      if (clientCall !== call) {
        // Don't send back to the sender
        clientCall.write(chatMessage);
      }
    });
  });

  // Handle client disconnecting or stream ending
  call.on("end", () => {
    console.log(`Client ${clientId} disconnected.`);
    // Remove the client's stream from the active list
    const index = connectedClients.indexOf(call);
    if (index > -1) {
      connectedClients.splice(index, 1);
    }
    call.end(); // End the server's side of the stream
  });

  call.on("error", (err) => {
    console.error(`Client ${clientId} stream error:`, err.message);
    const index = connectedClients.indexOf(call);
    if (index > -1) {
      connectedClients.splice(index, 1);
    }
  });

  call.on("status", (status) => {
    console.log(`Client ${clientId} stream status:`, status);
  });
}

/**
 * Starts an RPC server that receives requests for the ChatService service at the
 * sample server port
 */
function main() {
  const server = new grpc.Server();
  server.addService(chat_proto.ChatService.service, { chat: chat });
  server.bindAsync(
    "0.0.0.0:50051",
    grpc.ServerCredentials.createInsecure(),
    (err, port) => {
      if (err) {
        console.error("Server binding failed:", err);
        return;
      }
      server.start();
      console.log("gRPC server running on port 50051");
    }
  );
}

main();