Ejemplo GRPC

App.js

import React from 'react';
import logo from './logo.svg';
import './App.css';
const { PubSubClient } = require('./pubsub_grpc_web_pb');
const { SubRequest } = require('./pubsub_pb.js');

class App extends React.Component {

  constructor(props) {
    super(props)

    const pubSubService = new PubSubClient('http://localhost:9090');
    const request = new SubRequest();
    request.setName("topic");

    const stream = pubSubService.subscribe(request);

    stream.on('data', (response) => {
      console.log(response.getMessage());
    });

    stream.on('status', console.log);

    stream.on('end', (end) => {
      console.log("Stream end")
    });
  }

  render() {
    return (
      <div className="App">
        <header className="App-header">
          <img src={logo} className="App-logo" alt="logo" />
          <p>
            Edit <code>src/App.js</code> and save to reload.
          </p>
          <a
            className="App-link"
            href="https://reactjs.org"
            target="_blank"
            rel="noopener noreferrer"
          >
            Learn React
          </a>
        </header>
      </div>
    );
  }
}

export default App;

grpc_compile

#!/usr/bin/env bash

#Lo hice yo para compilar el proto en js

protoc -I=$(pwd)/src/grpc pubsub.proto \
  --js_out=import_style=commonjs:$(pwd)/src \
  --grpc-web_out=import_style=commonjs,mode=grpcwebtext:$(pwd)/src

cd src
echo $(PWD)
for F in $(ls *_pb.js)
do
  echo "Prepending file: $F"
  echo '/* eslint-disable */' | cat - $F > temp && mv temp $F
done

pubsub.proto

syntax = "proto3";

option java_multiple_files = true;
option java_package = "live";

package live;

service PubSub {
    rpc Subscribe (SubRequest) returns (stream Event) {}
}

message SubRequest {
    string name = 1;
}

message Event {
    string message = 1;
}

PubSubConsumer.scala

package live

import akka.actor.ActorSystem
import akka.grpc.GrpcClientSettings
import akka.stream.ActorMaterializer

import scala.concurrent.ExecutionContextExecutor
import scala.util.{Failure, Success}

object PubSubConsumer {

  def main(args: Array[String]): Unit = {
    implicit val sys: ActorSystem = ActorSystem("HelloWorldClient")
    implicit val mat: ActorMaterializer = ActorMaterializer()
    implicit val ec: ExecutionContextExecutor = sys.dispatcher

    val clientSettings = GrpcClientSettings.connectToServiceAt("127.0.0.1", 9090)
    val client: PubSub = PubSubClient(clientSettings)

    val respuestas = client.subscribe(SubRequest("topic"))

    val x = respuestas.runFold(0)((numero, evento) => {
      println(s"Msg #$numero: ${evento.message}")
      numero + 1
    })

    x.onComplete {
      case Success(_) =>
        println("stream finalizado")
      case Failure(e) =>
        println(s"Error: $e")
    }
  }
}

PubSubImpl.scala

package live

import akka.NotUsed
import akka.stream.Materializer
import akka.stream.scaladsl.Source

import scala.concurrent.duration._

class PubSubImpl(implicit mat: Materializer) extends PubSub {


  override def subscribe(in: SubRequest): Source[Event, NotUsed] = {
    println(s"SubRequest: ${in.name}")
    Source.repeat(Event("Hola!"))
      .throttle(1, 1.seconds)
  }
}

PubSubServer.scala

package live

/*
 * Copyright (C) 2018-2019 Lightbend Inc. <https://www.lightbend.com>
 */

//#full-server

import akka.actor.ActorSystem
import akka.http.scaladsl.model.{HttpRequest, HttpResponse}
import akka.http.scaladsl.{Http, HttpConnectionContext}
import akka.stream.{ActorMaterializer, Materializer}
import com.typesafe.config.ConfigFactory

import scala.concurrent.{ExecutionContext, Future}

object PubSubServer {

  def main(args: Array[String]): Unit = {
    val conf = ConfigFactory
      .parseString("akka.http.server.preview.enable-http2 = on")
      .withFallback(ConfigFactory.defaultApplication())
    val system = ActorSystem("HelloWorld", conf)
    new PubSubServer(system).run()
  }
}

class PubSubServer(system: ActorSystem) {

  def run(): Future[Http.ServerBinding] = {
    
    implicit val sys: ActorSystem = system
    implicit val mat: Materializer = ActorMaterializer()
    implicit val ec: ExecutionContext = sys.dispatcher

    val service: HttpRequest => Future[HttpResponse] =
      PubSubHandler(new PubSubImpl)
    
    val handler: HttpRequest => Future[HttpResponse] = { request =>
      val withoutEncoding = request.copy(headers = request.headers.filterNot(_.name == "grpc-accept-encoding"))
      service(withoutEncoding)
    }

    val binding = Http().bindAndHandleAsync(
      handler,
      interface = "127.0.0.1",
      port = 8080,
      connectionContext = HttpConnectionContext())

    binding.foreach { binding =>
      println(s"gRPC server bound to: ${binding.localAddress}")
    }

    binding
  }
}