सर्वर स्ट्रीमिंग की व्याख्या

syntax = "proto3";

option java_package = "com.coddykit.grpc";
option java_outer_classname = "StreamingProto";

service NotifierService {
  rpc SubscribeToNotifications (SubscriptionRequest) returns (stream Notification);
}

message SubscriptionRequest {
  string userId = 1;
}

message Notification {
  string message = 1;
  int64 timestamp = 2;
}

// Example (Java)
public void subscribeToNotifications(SubscriptionRequest request,
    io.grpc.stub.StreamObserver<Notification> responseObserver) {

  String userId = request.getUserId();
  System.out.println("Client " + userId + " subscribed.");

  // Simulate sending multiple notifications
  for (int i = 0; i < 3; i++) {
    Notification notification = Notification.newBuilder()
        .setMessage("Update " + (i + 1) + " for " + userId)
        .setTimestamp(System.currentTimeMillis())
        .build();
    responseObserver.onNext(notification); // Send a message
    try {
      Thread.sleep(1000); // Wait a bit
    } catch (InterruptedException e) { /* handle */ }
  }

  responseObserver.onCompleted(); // Signal completion
  System.out.println("Finished sending notifications to " + userId);
}

// Example (Java)
StreamObserver<Notification> responseObserver = new StreamObserver<Notification>() {
  @Override
  public void onNext(Notification notification) {
    System.out.println("Received: " + notification.getMessage());
  }

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

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

// Call the streaming method
// asyncStub.subscribeToNotifications(request, responseObserver);

import io.grpc.Server;
import io.grpc.ServerBuilder;
import io.grpc.stub.StreamObserver;
import com.coddykit.grpc.StreamingProto.SubscriptionRequest;
import com.coddykit.grpc.StreamingProto.Notification;
import com.coddykit.grpc.NotifierServiceGrpc.NotifierServiceImplBase;

public class StreamingServer {

  private Server server;

  private void start() throws Exception {
    int port = 50051;
    server = ServerBuilder.forPort(port)
        .addService(new NotifierServiceImpl())
        .build()
        .start();
    System.out.println("Server started, listening on " + port);

    Runtime.getRuntime().addShutdownHook(new Thread() {
      @Override
      public void run() {
        System.err.println("*** shutting down gRPC server since JVM is shutting down");
        StreamingServer.this.stop();
        System.err.println("*** server shut down");
      }
    });
  }

  private void stop() {
    if (server != null) {
      server.shutdown();
    }
  }

  private void blockUntilShutdown() throws InterruptedException {
    if (server != null) {
      server.awaitTermination();
    }
  }

  public static void main(String[] args) throws Exception {
    final StreamingServer server = new StreamingServer();
    server.start();
    server.blockUntilShutdown();
  }

  static class NotifierServiceImpl extends NotifierServiceImplBase {
    @Override
    public void subscribeToNotifications(SubscriptionRequest request,
        StreamObserver<Notification> responseObserver) {

      String userId = request.getUserId();
      System.out.println("Server received subscription from: " + userId);

      for (int i = 0; i < 3; i++) {
        Notification notification = Notification.newBuilder()
            .setMessage("Update " + (i + 1) + " for " + userId)
            .setTimestamp(System.currentTimeMillis())
            .build();
        responseObserver.onNext(notification);
        try {
          Thread.sleep(1000); // Simulate some work
        } catch (InterruptedException e) {
          Thread.currentThread().interrupt();
          responseObserver.onError(e);
          return;
        }
      }
      responseObserver.onCompleted();
      System.out.println("Server finished sending notifications to: " + userId);
    }
  }
}

import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.StreamObserver;
import com.coddykit.grpc.StreamingProto.SubscriptionRequest;
import com.coddykit.grpc.StreamingProto.Notification;
import com.coddykit.grpc.NotifierServiceGrpc;

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

public class StreamingClient {

  public static void main(String[] args) throws InterruptedException {
    ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 50051)
        .usePlaintext() // For local testing, no TLS
        .build();

    NotifierServiceGrpc.Stub asyncStub = NotifierServiceGrpc.newStub(channel);

    CountDownLatch latch = new CountDownLatch(1);

    System.out.println("Client sending subscription request...");
    SubscriptionRequest request = SubscriptionRequest.newBuilder()
        .setUserId("user123")
        .build();

    asyncStub.subscribeToNotifications(request, new StreamObserver<Notification>() {
      @Override
      public void onNext(Notification notification) {
        System.out.println("Client received notification: " + notification.getMessage());
      }

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

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

    latch.await(5, TimeUnit.SECONDS); // Wait for stream to complete
    System.out.println("Client finished.");

    channel.shutdown().awaitTermination(5, TimeUnit.SECONDS);
  }
}