Streaming Dua Arah

syntax = "proto3";

package chat;

message ChatMessage {
  string user = 1;
  string message = 2;
}

service ChatService {
  rpc Chat(stream ChatMessage) returns (stream ChatMessage);
}

import io.grpc.Server;
import io.grpc.ServerBuilder;
import io.grpc.stub.StreamObserver;

// Assuming ChatServiceGrpc and ChatMessage are generated
// from the proto definition.
import chat.ChatServiceGrpc;
import chat.ChatMessage;

public class ChatServer {

  public static void main(String[] args) throws Exception {
    Server server = ServerBuilder.forPort(50051)
        .addService(new ChatServiceImpl())
        .build();

    server.start();
    System.out.println("Server started on port 50051");
    server.awaitTermination();
  }

  static class ChatServiceImpl extends ChatServiceGrpc.ChatServiceImplBase {
    @Override
    public StreamObserver<ChatMessage> chat(
        final StreamObserver<ChatMessage> responseObserver) {

      return new StreamObserver<ChatMessage>() {
        @Override
        public void onNext(ChatMessage request) {
          // Received a message from the client
          System.out.println("Server received: " + request.getMessage());
          // Echo it back to the client
          ChatMessage response = ChatMessage.newBuilder()
              .setUser("Server")
              .setMessage("Echo: " + request.getMessage())
              .build();
          responseObserver.onNext(response);
        }

        @Override
        public void onError(Throwable t) {
          System.err.println("Server error: " + t.getMessage());
        }

        @Override
        public void onCompleted() {
          System.out.println("Client stream completed.");
          responseObserver.onCompleted(); // Close server stream
        }
      };
    }
  }
}

import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.StreamObserver;

// Assuming ChatServiceGrpc and ChatMessage are generated
// from the proto definition.
import chat.ChatServiceGrpc;
import chat.ChatMessage;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

public class ChatClient {

  public static void main(String[] args) throws Exception {
    ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 50051)
        .usePlaintext() // For demonstration, use TLS in production
        .build();

    ChatServiceGrpc.ChatStub asyncStub = ChatServiceGrpc.newStub(channel);

    CountDownLatch finishLatch = new CountDownLatch(1);

    StreamObserver<ChatMessage> requestObserver = asyncStub.chat(
        new StreamObserver<ChatMessage>() {
          @Override
          public void onNext(ChatMessage response) {
            System.out.println("Client received: " + response.getMessage());
          }

          @Override
          public void onError(Throwable t) {
            System.err.println("Client error: " + t.getMessage());
            finishLatch.countDown();
          }

          @Override
          public void onCompleted() {
            System.out.println("Server stream completed.");
            finishLatch.countDown();
          }
        });

    try {
      // Client sends messages
      for (int i = 0; i < 3; i++) {
        ChatMessage request = ChatMessage.newBuilder()
            .setUser("Client")
            .setMessage("Hello " + i)
            .build();
        requestObserver.onNext(request);
        Thread.sleep(500); // Simulate delay
      }
    } catch (RuntimeException | InterruptedException e) {
      requestObserver.onError(e);
      throw e;
    } finally {
      requestObserver.onCompleted(); // Client finishes sending
    }

    if (!finishLatch.await(1, TimeUnit.MINUTES)) {
      System.err.println("Client timed out waiting for server response.");
    }
    channel.shutdownNow().awaitTermination(5, TimeUnit.SECONDS);
  }
}