Need an help with implementation of ZIO-GRPC

#zio #scala #grpc

Also refer this stack overflow: https://stackoverflow.com/questions/79426489/need-help-implementing-grpc-pub-sub-using-zio
I am able to connect GRPC server using scala and fetchResponse by calling the subscribe api using grpc StreamObserver. With the same configuration I tried using ZIO framework I am not reciving the response.

I have below configuration host: api.salesforce.pubsub.com port : 7443 tenantId : "xxx" accessToken : "yyy" instanceUrl : "https://zzz.com"

I tried with this configuration to build ManagedChannel and call subscribe Api by passing FetchRequest(topicName) i am successfully getting the FetchResponse using StreamObserver.

With the same configuration I am not able to receive the FetchResponse and not even getting an error. Below is the ZIO-GRPC implementation.

object Main extends ZIOAppDefault{
def managedChannel():ZManagedChannel={
val managedChannelBuilder: ManagedChannelBuilder[_] = ManagedChannelBuilder.forAddress(host,port)

val interceptor = Seq(ZClientInterceptor.headersReplacer{
  (_,_)=> SafeMetadata.make(
       ("tenantid",tenantId),
       ("accesstoken",accessToken),
       ("instanceurl",URI(instanceUrl).resolve("/").toString))
  })

ZManagedChannel(managedChannelBuilder, interceptor)
}

def requestBody() ={
val fetchRequest: zio.stream.Stream[StatusException,FetchRequest]=ZStream.succeed(FetchRequest(topicName= "topic", replayPresent = EARLIEST))

fetchRequest

}

def subcribeAndFetch():ZStream[PubSubClient, StatusException,Option[FetchResponse]]=
for{
_ <- ZStream.logInfo(s"fetching..")
response <- PubSubClient.subscribe(requestBody()).map(response => Some(response))
.catchAll(ex => ZStream.logError(s"the subscribe request failed: ${ex.getStatus}").as(None))
} yield response

override def run ={
subcribeAndFetch
.runDrain
.repeat(Schedule.spaced(2.second))
.provideLayer(PubSubClient.live(managedChannel()))
}
}