service JDBCServices {rpc greeting(stream HelloMsg) returns (stream HelloMsg) {}
}service CQLServices {rpc greeting(stream HelloMsg) returns (stream HelloMsg) {}
}service MGOServices {rpc greeting(stream HelloMsg) returns (stream HelloMsg) {}
}

syntax = "proto3";package sdp.grpc.services;message HelloMsg {string hello = 1;
}message DataRow {string countyname = 1;string statename = 2;int32 reportyear = 3;int32 value = 4;
}

syntax = "proto3";import "google/protobuf/wrappers.proto";
import "google/protobuf/any.proto";
import "scalapb/scalapb.proto";option (scalapb.options) = {// use a custom Scala package name// package_name: "io.ontherocks.introgrpc.demo"// don't append file name to packageflat_package: true// generate one Scala file for all messages (services still get their own file)single_file: true// add imports to generated file// useful when extending traits or using custom types// import: "io.ontherocks.hellogrpc.RockingMessage"// code to put at the top of generated file// works only with `single_file: true`//preamble: "sealed trait SomeSealedTrait"
};/** Demoes various customization options provided by ScalaPBs.*/package sdp.grpc.services;import "misc/sdp.proto";
import "common.proto";
import "cql/cql.proto";
import "jdbc/jdbc.proto";
import "mgo/mgo.proto";

import sdp.grpc.mongo.client.MGOClientclass CQLStreamingServices(implicit ec: ExecutionContextExecutor,mat: ActorMaterializer,  session: Session)extends CqlGrpcAkkaStream.CQLServices with LogSupport{val mongoClient = new MGOClientval stub = mongoClient.stubdef sayHelloTo(msg: String): Flow[HelloMsg, HelloMsg, NotUsed] =Flow[HelloMsg].map { r => HelloMsg(r.hello + msg)}.via(stub.greeting)override def greeting: Flow[HelloMsg, HelloMsg, NotUsed] =Flow[HelloMsg].via(sayHelloTo(",from cassandra to mongo"))}

package sdp.grpc.mongo.clientimport sdp.grpc.services._
import sdp.logging.LogSupport
import io.grpc._
import common._
import sdp.grpc.services._
import akka.stream.scaladsl._
import akka.NotUsedclass MGOClient extends LogSupport {val channel = ManagedChannelBuilder.forAddress("localhost", 50051).usePlaintext().build()val stub = MgoGrpcAkkaStream.stub(channel)}

import sdp.grpc.cassandra.client.CQLClientclass JDBCStreamingServices(implicit ec: ExecutionContextExecutor)extends JdbcGrpcAkkaStream.JDBCServices with LogSupport {val cassandraClient = new CQLClientval stub = cassandraClient.stubdef sayHelloTo(msg: String): Flow[HelloMsg,HelloMsg,NotUsed] =Flow[HelloMsg].map {r => HelloMsg(r.hello + msg)}.via(stub.greeting)override def greeting: Flow[HelloMsg, HelloMsg, NotUsed] =Flow[HelloMsg].via(sayHelloTo(",from jdbc to cassandra"))}

package demo.sdp.grpcimport akka.actor.ActorSystem
import akka.stream.{ActorMaterializer, ThrottleMode}import sdp.grpc.jdbc.client.JDBCClientobject DemoApp extends App {implicit val system = ActorSystem("jdbcClient")implicit val mat = ActorMaterializer.create(system)implicit val ec = system.dispatcherval jdbcClient = new JDBCClientjdbcClient.sayHello.runForeach(r => println(r.hello))scala.io.StdIn.readLine()mat.shutdown()system.terminate()}

package sdp.grpc.jdbc.clientimport sdp.grpc.services._
import sdp.logging.LogSupport
import io.grpc._
import common._
import sdp.grpc.services._
import akka.stream.scaladsl._
import akka.NotUsedclass JDBCClient extends LogSupport {val channel = ManagedChannelBuilder.forAddress("localhost", 50053).usePlaintext().build()val stub = JdbcGrpcAkkaStream.stub(channel)def sayHello: Source[HelloMsg, NotUsed] = {val row = HelloMsg("hello ")val rows = List.fill[HelloMsg](100)(row)Source.fromIterator(() => rows.iterator).via(stub.greeting)}
}

hello ,from jdbc to cassandra,from cassandra to mongo, mongo says hi
hello ,from jdbc to cassandra,from cassandra to mongo, mongo says hi
hello ,from jdbc to cassandra,from cassandra to mongo, mongo says hi
hello ,from jdbc to cassandra,from cassandra to mongo, mongo says hi
hello ,from jdbc to cassandra,from cassandra to mongo, mongo says hi
...