:spark: #25 use async grpc client to call

Author: jcai

wechaty-puppet-hostie/src/main/scala/wechaty/hostie/support/ContactRawSupport.scala

@@ -9,6 +9,8 @@ import wechaty.puppet.schemas.Contact.ContactPayload
 import wechaty.puppet.schemas.Puppet
 import wechaty.puppet.support.ContactSupport
 
+import scala.concurrent.Future
+
 /**
   *
   * @author <a href="mailto:[email protected]">Jun Tsai</a>
@@ -80,23 +82,24 @@ trait ContactRawSupport {
     file
   }
 
-  override protected def contactRawPayload(contactID: String): ContactPayload = {
-    val response = grpcClient.contactPayload(ContactPayloadRequest.newBuilder().setId(contactID).build())
-    val contact = new ContactPayload
-    contact.id = response.getId
-    contact.gender = wechaty.puppet.schemas.Contact.ContactGender.apply(response.getGenderValue)
-    contact.`type` = wechaty.puppet.schemas.Contact.ContactType.apply(response.getTypeValue)
-    contact.name = response.getName
-    contact.avatar = response.getAvatar
-    contact.address = response.getAddress
-    contact.alias = response.getAlias
-    contact.city = response.getCity
-    contact.friend = response.getFriend
-    contact.province = response.getProvince
-    contact.signature = response.getSignature
-    contact.star = response.getStar
-    contact.weixin = response.getWeixin
-
-    contact
+  override protected def contactRawPayload(contactID: String): Future[ContactPayload] = {
+    val request = ContactPayloadRequest.newBuilder().setId(contactID).build()
+    asyncCallback(asyncGrpcClient.contactPayload,request){response =>
+      val contact = new ContactPayload
+      contact.id = response.getId
+      contact.gender = wechaty.puppet.schemas.Contact.ContactGender.apply(response.getGenderValue)
+      contact.`type` = wechaty.puppet.schemas.Contact.ContactType.apply(response.getTypeValue)
+      contact.name = response.getName
+      contact.avatar = response.getAvatar
+      contact.address = response.getAddress
+      contact.alias = response.getAlias
+      contact.city = response.getCity
+      contact.friend = response.getFriend
+      contact.province = response.getProvince
+      contact.signature = response.getSignature
+      contact.star = response.getStar
+      contact.weixin = response.getWeixin
+      contact
+    }
   }
 }

wechaty-puppet-hostie/src/main/scala/wechaty/hostie/support/GrpcSupport.scala

@@ -5,11 +5,16 @@ import java.util.concurrent.{CountDownLatch, Executors, TimeUnit}
 
 import com.typesafe.scalalogging.LazyLogging
 import io.github.wechaty.grpc.PuppetGrpc
-import io.github.wechaty.grpc.puppet.Base
+import io.github.wechaty.grpc.puppet.Contact.ContactPayloadRequest
+import io.github.wechaty.grpc.puppet.{Base, Contact}
 import io.github.wechaty.grpc.puppet.Event.EventRequest
+import io.grpc.stub.ClientCalls.asyncUnaryCall
 import io.grpc.stub.StreamObserver
-import io.grpc.{ManagedChannel, ManagedChannelBuilder}
+import io.grpc.{ClientCall, ManagedChannel, ManagedChannelBuilder}
 import wechaty.hostie.PuppetHostie
+import wechaty.puppet.schemas.Contact.ContactPayload
+
+import scala.concurrent.{Future, Promise}
 
 /**
   *
@@ -23,9 +28,9 @@ trait GrpcSupport {
   private val HEARTBEAT_COUNTER = new AtomicLong()
   private val HOSTIE_KEEPALIVE_TIMEOUT = 15 * 1000L
   private val DEFAULT_WATCHDOG_TIMEOUT = 60L
-  protected var grpcClient: PuppetGrpc.PuppetBlockingStub = _
-  private var eventStream: PuppetGrpc.PuppetStub = _
-  protected var channel: ManagedChannel = _
+  protected var grpcClient     : PuppetGrpc.PuppetBlockingStub = _
+  protected var asyncGrpcClient: PuppetGrpc.PuppetStub         = _
+  protected var channel        : ManagedChannel                = _
 
   protected def startGrpc(endpoint: String): Unit = {
     initChannel(endpoint)
@@ -96,9 +101,9 @@ trait GrpcSupport {
   }
 
   private def startStream() {
-    this.eventStream = PuppetGrpc.newStub(channel)
+    this.asyncGrpcClient = PuppetGrpc.newStub(channel)
     val startRequest = EventRequest.newBuilder().build()
-    this.eventStream.event(startRequest, this)
+    this.asyncGrpcClient.event(startRequest, this)
   }
 
   protected def stopGrpc(): Unit = {
@@ -120,7 +125,7 @@ trait GrpcSupport {
   private def stopStream(): Unit = {
     try {
       val countDownLatch = new CountDownLatch(1)
-      this.eventStream.stop(Base.StopRequest.getDefaultInstance, new StreamObserver[Base.StopResponse] {
+      this.asyncGrpcClient.stop(Base.StopRequest.getDefaultInstance, new StreamObserver[Base.StopResponse] {
         override def onNext(v: Base.StopResponse): Unit = {}
 
         override def onError(throwable: Throwable): Unit = {}
@@ -133,4 +138,24 @@ trait GrpcSupport {
         logger.warn("fail to stop stream {}", e.getMessage)
     }
   }
+
+  type ClientCall[ReqT,RespT]=(ReqT,StreamObserver[RespT])=>Unit
+  type ClientCallback[RespT,T]=RespT => T
+  protected def asyncCall[ReqT,RespT](call: ClientCall[ReqT, RespT], req: ReqT): Unit = {
+    asyncCallback(call,req)(resp=> resp)
+  }
+  protected def asyncCallback[ReqT,RespT,T](call: ClientCall[ReqT, RespT], req: ReqT)(callback:ClientCallback[RespT,T]):Future[T]= {
+    val promise = Promise[T]
+    call(req,new StreamObserver[RespT] {
+      override def onNext(value: RespT): Unit = {
+        val result = callback(value)
+        promise.success(result)
+      }
+      override def onError(t: Throwable): Unit = promise.failure(t)
+      override def onCompleted(): Unit = {
+        if(!promise.isCompleted) promise.failure(new IllegalStateException("server completed"))
+      }
+    })
+    promise.future
+  }
 }

wechaty-puppet-padplus/src/main/scala/wechaty/padplus/support/ContactRawSupport.scala

@@ -50,17 +50,17 @@ trait ContactRawSupport {
   /**
     * contact
     */
-  override protected def contactRawPayload(contactId: String): Contact.ContactPayload = {
-    this.getContact(contactId)
+  override protected def contactRawPayload(contactId: String): Future[Contact.ContactPayload] = {
+    this.getContact(contactId).map(convertPadplusContactToContactPayload)
   }
 
-  protected def getContact(contactId: String): PadplusContactPayload = {
+  protected def getContact(contactId: String): Future[PadplusContactPayload] = {
     getPadplusContactPayload(contactId) match{
-      case Some(padplusContactPayload) => padplusContactPayload
+      case Some(padplusContactPayload) => Future.successful(padplusContactPayload)
       case _ =>
         val json = objectMapper.createObjectNode()
         json.put("userName", contactId)
-        syncRequest[GrpcContactPayload](ApiType.GET_CONTACT,Some(json.toString))
+        asyncRequest[GrpcContactPayload](ApiType.GET_CONTACT,Some(json.toString)).map(convertFromGrpcContact)
     }
   }
 

wechaty-puppet-padplus/src/test/scala/wechaty/padplus/support/ContactRawSupportTest.scala

@@ -35,7 +35,7 @@ class ContactRawSupportTest extends PadplusTestEventBase{
     val payload = Await.result(future,10 seconds)
     Assertions.assertEquals("jcai",payload.id)
     awaitEventCompletion()
-    
+
      */
   }
 }

wechaty-puppet/src/main/scala/wechaty/puppet/schemas/Puppet.scala

@@ -7,20 +7,23 @@ import com.fasterxml.jackson.module.scala.DefaultScalaModule
 import io.github.wechaty.grpc.puppet.Event.EventType
 import io.grpc.ManagedChannel
 
+import scala.concurrent.{ExecutionContext, ExecutionContextExecutor}
+
 /**
   *
   * @author <a href="mailto:[email protected]">Jun Tsai</a>
   * @since 2020-06-01
   */
 object Puppet {
+  implicit lazy val executionContext: ExecutionContextExecutor =  ExecutionContext.global
   class PuppetOptions {
     var endPoint: Option[String] = None
     var timeout: Option[Long] = None
     var token: Option[String] = None
     var channelOpt:Option[ManagedChannel] = None
     var puppetOptionKey: Option[String] = None
   }
-  lazy val objectMapper = {
+  lazy val objectMapper: ObjectMapper = {
     val om = new ObjectMapper()
     import com.fasterxml.jackson.databind.DeserializationFeature
     om.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)

wechaty-puppet/src/main/scala/wechaty/puppet/support/ContactSupport.scala

@@ -5,6 +5,9 @@ import com.typesafe.scalalogging.LazyLogging
 import wechaty.puppet.ResourceBox
 import wechaty.puppet.schemas.Contact.ContactPayload
 import wechaty.puppet.schemas.Puppet
+import wechaty.puppet.schemas.Puppet.executionContext
+
+import scala.concurrent.Future
 
 /**
   *
@@ -29,7 +32,7 @@ trait ContactSupport {
 
   def contactList(): Array[String]
 
-  def contactPayload(contactId: String): ContactPayload = {
+  def contactPayload(contactId: String): Future[ContactPayload] = {
     if (Puppet.isBlank(contactId)) {
       throw new IllegalArgumentException("contact id is blank!")
     }
@@ -39,18 +42,19 @@ trait ContactSupport {
       */
     val cachedPayload = this.cacheContactPayload.getIfPresent(contactId)
     if (cachedPayload != null) {
-      return cachedPayload
+      return Future.successful(cachedPayload)
     }
 
     /**
       * 2. Cache not found
       */
-    val payload = this.contactRawPayload(contactId)
-
-    cacheContactPayload.put(contactId, payload)
-    logger.info("Puppet contactPayload({}) cache SET", contactId)
+    val payloadFuture = this.contactRawPayload(contactId)
 
-    payload
+    payloadFuture.map(payload=>{
+      cacheContactPayload.put(contactId, payload)
+      logger.info("Puppet contactPayload({}) cache SET", contactId)
+      payload
+    })
   }
   def contactPayloadDirty (contactId: String): Unit ={
     logger.debug("Puppet contactPayloadDirty({})", contactId)
@@ -60,6 +64,6 @@ trait ContactSupport {
   /**
     * contact
     */
-  protected def contactRawPayload(contactId: String): ContactPayload
+  protected def contactRawPayload(contactId: String): Future[ContactPayload]
 
 }

wechaty/src/main/scala/wechaty/user/Contact.scala

@@ -6,6 +6,8 @@ import wechaty.puppet.ResourceBox
 import wechaty.puppet.schemas.Contact.{ContactGender, ContactPayload, ContactType}
 import wechaty.puppet.schemas.Puppet
 
+import scala.concurrent.Await
+import scala.concurrent.duration._
 import scala.language.implicitConversions
 
 /**
@@ -21,7 +23,9 @@ import scala.language.implicitConversions
 class Contact(contactId: String)(implicit resolver: PuppetResolver) extends Conversation(contactId) with LazyLogging {
   //  lazy val payload: schemas.Contact.ContactPayload = resolver.puppet.contactPayload(contactId)
   def payload: ContactPayload = {
-    resolver.puppet.contactPayload(contactId)
+    //TODO use async
+    val f = resolver.puppet.contactPayload(contactId)
+    Await.result(f,10 seconds)
   }