#21 #25 optimize async call method

Author: jcai

wechaty-puppet-padplus/src/main/resources/log4j.properties

@@ -8,4 +8,4 @@ log4j.appender.A1=org.apache.log4j.ConsoleAppender
 # A1 uses PatternLayout.
 log4j.appender.A1.layout=org.apache.log4j.PatternLayout
 log4j.appender.A1.layout.ConversionPattern=[%t %p]%d{HH:mm:ss}  %c{2} %m%n
-log4j.category.wechaty=debug
+log4j.category.wechaty=trace

wechaty-puppet-padplus/src/main/scala/wechaty/padplus/PuppetPadplus.scala

@@ -46,9 +46,9 @@ class PuppetPadplus(val option:PuppetOptions,val storePath:String="/tmp/padplus"
       case Some(str) =>
         uinOpt = Some(str)
         logger.debug("found uin in local store:{}",str)
-        asyncRequest(ApiType.INIT)
+        asyncRequestNothing(ApiType.INIT)
       case _ =>
-        asyncRequest(ApiType.GET_QRCODE)
+        asyncRequestNothing(ApiType.GET_QRCODE)
     }
   }
   def stop(): Unit = {

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

@@ -4,9 +4,12 @@ import java.util.concurrent.TimeUnit
 
 import com.github.benmanes.caffeine.cache.{Cache, Caffeine}
 import wechaty.padplus.grpc.PadPlusServerOuterClass.StreamResponse
-import wechaty.padplus.schemas.ModelContact.{PadplusContactPayload, PadplusConversation}
+import wechaty.padplus.schemas.ModelContact.PadplusContactPayload
 import wechaty.padplus.schemas.ModelRoom.{PadplusRoomMemberMap, PadplusRoomPayload}
 
+import scala.concurrent.Promise
+import scala.util.{Success, Try}
+
 /**
   *
   * @author <a href="mailto:[email protected]">Jun Tsai</a>
@@ -19,23 +22,25 @@ object CallbackHelper {
       .build()
       .asInstanceOf[Cache[String, T]]
   }
-  type TraceRequestCallback = StreamResponse => Unit
+  type TraceRequestCallback = Promise[StreamResponse]
   lazy val traceCallbacks: Cache[String, TraceRequestCallback] = createPool[TraceRequestCallback]
 
-  type ContactCallback = PadplusConversation => Unit
-  lazy val contactCallbacks: Cache[String, List[ContactCallback]] = createPool[List[ContactCallback]]
+  type ContactPromise= Promise[PadplusContactPayload]
+  lazy val contactCallbacks: Cache[String, List[ContactPromise]] = createPool[List[ContactPromise]]
+  type RoomPromise= Promise[PadplusRoomPayload]
+  lazy val roomCallbacks: Cache[String, List[RoomPromise]] = createPool[List[RoomPromise]]
 
-  type ContactAliasCallback = () =>  Unit
+  type ContactAliasCallback = Promise[Unit]
   lazy val contactAliasCallbacks: Cache[String, Map[String, ContactAliasCallback]] = createPool[Map[String,ContactAliasCallback]]
 
 
-  type RoomTopicCallback= ()=>Unit
+  type RoomTopicCallback= Promise[Unit]
   lazy val roomTopicCallbacks: Cache[String, Map[String, RoomTopicCallback]] = createPool[Map[String,RoomTopicCallback]]
 
-  type AcceptFriendCallback= () =>Unit
+  type AcceptFriendCallback= Promise[Unit]
   lazy val acceptFriendCallbacks: Cache[String, List[AcceptFriendCallback]] = createPool[List[AcceptFriendCallback]]
 
-  type RoomMemberCallback= PadplusRoomMemberMap =>Unit
+  type RoomMemberCallback= Promise[PadplusRoomMemberMap]
   lazy val roomMemberCallbacks: Cache[String, List[RoomMemberCallback]] = createPool[List[RoomMemberCallback]]
 
   def pushCallbackToPool (traceId: String, callback: TraceRequestCallback){
@@ -44,7 +49,7 @@ object CallbackHelper {
   def resolveCallBack (traceId: String,response:StreamResponse):Unit={
     val callback = traceCallbacks.getIfPresent(traceId)
     if(callback!=null){
-      callback(response)
+      callback.success(response)
     }
     traceCallbacks.invalidate(traceId)
   }
@@ -61,32 +66,42 @@ object CallbackHelper {
         cache.put(key,callbackList :+ callback)
       }
   }
+  private def resolveListCallBack[T] (cache:Cache[String,List[Promise[T]]],key: String, data: Try[T]) {
+    val callbackList = cache.getIfPresent(key)
+    if (callbackList != null) {
+      callbackList.foreach(f => f.complete(data))
+      cache.invalidate(key)
+    }
+  }
 
-  def pushContactCallback ( contactId: String, callback: ContactCallback) {
+  def pushContactCallback ( contactId: String, callback:ContactPromise) {
     pushListCallback(contactCallbacks,contactId,callback)
   }
 
-  def resolveContactCallBack (contactId: String, data: PadplusContactPayload) {
-    val callbackList = contactCallbacks.getIfPresent(contactId)
-    if(callbackList != null){
-      callbackList.foreach(f=>f(data))
-    }
+  def resolveContactCallBack (contactId: String, data: Try[PadplusContactPayload]) {
+    resolveListCallBack(contactCallbacks,contactId,data)
 
-    this.resolveContactAliasCallback(contactId, data.remark)
-    this.resolveAcceptFriendCallback(contactId)
-    contactCallbacks.invalidate(contactId)
+    data match{
+      case Success(value) =>
+        this.resolveContactAliasCallback(contactId, value.remark)
+        this.resolveAcceptFriendCallback(contactId)
+      case _ =>
+    }
   }
 
-  def resolveRoomCallBack (roomId: String, data: PadplusRoomPayload) {
-    val callbackList = contactCallbacks.getIfPresent(roomId)
-    if(callbackList != null){
-      callbackList.foreach(f=>f(data))
+  def pushRoomCallback ( roomId: String, callback:RoomPromise) {
+    pushListCallback(roomCallbacks,roomId,callback)
+  }
+  def resolveRoomCallBack (roomId: String, data: Try[PadplusRoomPayload]) {
+    resolveListCallBack(roomCallbacks,roomId,data)
+    data match{
+      case Success(value) =>
+        this.resolveRoomTopicCallback(value.chatroomId, value.nickName)
+      case _ => //do nothing
     }
-    this.resolveRoomTopicCallback(data.chatroomId, data.nickName)
-    contactCallbacks.invalidate(roomId)
   }
 
-  private def pushMapCallback[T](cache:Cache[String,Map[String,T]],key:String,mapKey:String,callback:T): Unit ={
+  private def pushMapCallback[T](cache:Cache[String,Map[String,Promise[T]]],key:String,mapKey:String,callback:Promise[T]): Unit ={
     val callbacks = cache.getIfPresent(key)
     if(callbacks == null){
       cache.put(key,Map(mapKey->callback))
@@ -97,12 +112,12 @@ object CallbackHelper {
   def pushContactAliasCallback (contactId: String, alias: String, callback:ContactAliasCallback): Unit = {
     pushMapCallback(contactAliasCallbacks,contactId,alias,callback)
   }
-  private def resolveMapCallback(cache:Cache[String,Map[String,()=>Unit]],key: String, mapKey: String) {
+  private def resolveMapCallback(cache:Cache[String,Map[String,Promise[Unit]]],key: String, mapKey: String) {
     val callbacks = cache.getIfPresent(key)
     if(callbacks != null){
       val callbackOpt = callbacks.get(mapKey)
       callbackOpt.foreach{f=>
-        f()
+        f.success({})
         if(callbacks.size == 1)
           cache.invalidate(key)
         else
@@ -127,22 +142,14 @@ object CallbackHelper {
   }
 
   private def resolveAcceptFriendCallback (contactId: String) {
-    val callbacks = acceptFriendCallbacks.getIfPresent(contactId)
-    if(callbacks != null){
-      callbacks.foreach(cb=>cb())
-      acceptFriendCallbacks.invalidate(contactId)
-    }
+    resolveListCallBack(acceptFriendCallbacks,contactId,Try[Unit]({}))
   }
 
   def pushRoomMemberCallback (roomId: String, callback: RoomMemberCallback): Unit ={
     pushListCallback(roomMemberCallbacks,roomId,callback)
   }
 
-  def resolveRoomMemberCallback (roomId: String, memberList: PadplusRoomMemberMap) {
-    val callbacks = roomMemberCallbacks.getIfPresent(roomId)
-    if(callbacks != null){
-      callbacks.foreach(cb=>cb(memberList))
-      roomMemberCallbacks.invalidate(roomId)
-    }
+  def resolveRoomMemberCallback (roomId: String, memberList: Try[PadplusRoomMemberMap]) {
+    resolveListCallBack(roomMemberCallbacks,roomId,memberList)
   }
 }

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

@@ -21,7 +21,7 @@ import wechaty.puppet.schemas.Event.{EventLoginPayload, EventScanPayload}
 import wechaty.puppet.schemas.Puppet._
 import wechaty.puppet.schemas.{Contact, Puppet}
 
-import scala.concurrent.Future
+import scala.concurrent.{Future, Promise}
 import scala.language.implicitConversions
 import scala.util.Try
 
@@ -54,9 +54,11 @@ trait ContactRawSupport extends LazyLogging{
     getPadplusContactPayload(contactId) match{
       case Some(padplusContactPayload) => Future.successful(padplusContactPayload)
       case _ =>
+        val promise = Promise[PadplusContactPayload]
+        CallbackHelper.pushContactCallback(contactId,promise)
         val json = objectMapper.createObjectNode()
         json.put("userName", contactId)
-        asyncRequest[GrpcContactPayload](ApiType.GET_CONTACT,Some(json.toString)).map(convertFromGrpcContact)
+        asyncRequestNothing(ApiType.GET_CONTACT,Some(json.toString)).flatMap(_=>promise.future)
     }
   }
 
@@ -138,6 +140,7 @@ trait ContactRawSupport extends LazyLogging{
       if(!isBlank(data)){
         val root = objectMapper.readTree(data)
         val userName = root.get("UserName").asText()
+        logger.trace("response data:{}",data)
         if(isRoomId(userName)){
           val roomTry=Try {
             val grpcRoomPayload = objectMapper.readValue(data, classOf[GrpcRoomPayload])
@@ -157,9 +160,7 @@ trait ContactRawSupport extends LazyLogging{
 
             roomPayload
           }
-          //TODO use Try instance to avoid memory leak
-          if(roomTry.isSuccess)
-            CallbackHelper.resolveRoomCallBack(userName,roomTry.get)
+          CallbackHelper.resolveRoomCallBack(userName,roomTry)
 
         }else{
           val result:Try[PadplusContactPayload] = Try {
@@ -169,9 +170,7 @@ trait ContactRawSupport extends LazyLogging{
             padplusContactPayload
           }
 
-          //TODO use Try instance to avoid memory leak
-          if(result.isSuccess)
-            CallbackHelper.resolveContactCallBack(userName,result.get)
+          CallbackHelper.resolveContactCallBack(userName,result)
         }
       }
   }

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

@@ -126,7 +126,38 @@ trait GrpcSupport {
   protected def generateTraceId(apiType:ApiType): String={
     UUID.randomUUID().toString
   }
+  //can't create Promise[Nothing] instance,so use the method create Future[Unit]
+  protected def asyncRequestNothing(apiType: ApiType,data:Option[Any]=None): Future[Unit] ={
+    val request = RequestObject.newBuilder()
+    request.setToken(option.token.get)
+    uinOpt match{
+      case Some(id) =>
+        request.setUin(id)
+      case _ =>
+    }
+    request.setApiType(apiType)
+    data match{
+      case Some(str:String) =>
+        request.setParams(str)
+      case Some(d) =>
+        request.setParams(Puppet.objectMapper.writeValueAsString(d))
+      case _ =>
+    }
+
+    val future = asyncCall(PadPlusServerGrpc.getRequestMethod,request.build())
+    future.map{rep=>
+      if(rep.getResult != "success"){
+        logger.warn("fail to request {}",rep)
+        throw new IllegalAccessException("fail to request ,grpc result:"+rep)
+      }
+    }
+  }
   protected def asyncRequest[T : TypeTag ](apiType: ApiType,data:Option[Any]=None)(implicit classTag: ClassTag[T]): Future[T] ={
+    typeOf[T] match {
+      case t if t =:= typeOf[Nothing] =>
+        throw new IllegalAccessException("generic type is nothing,maybe you should use asyncRequestNothing !")
+      case _ =>
+    }
     val request = RequestObject.newBuilder()
     request.setToken(option.token.get)
     uinOpt match{
@@ -147,53 +178,31 @@ trait GrpcSupport {
     val traceId= generateTraceId(apiType)
     request.setTraceId(traceId)
     logger.debug("request:{}",request.build())
-    val p = Promise[T]()
-
-    val callbackDelegate:TraceRequestCallback=(streamResponse:StreamResponse)=>{
-      if(p.isCompleted){
-        logger.warn("promise is completed ,{}",p)
-      }else {
-        p.complete(Try {
-          typeOf[T] match{
-            case t if t =:= typeOf[Nothing] =>
-              null.asInstanceOf[T]
-            case t if t =:= typeOf[JsonNode] =>
-              Puppet.objectMapper.readTree(streamResponse.getData).asInstanceOf[T]
-            case _ =>
-              try {
-                Puppet.objectMapper.readValue(streamResponse.getData, classTag.runtimeClass).asInstanceOf[T]
-              }catch{
-                case e:Throwable =>
-                  logger.error(e.getMessage,e)
-                  throw e
-              }
-          }
-        })
-      }
-    }
-    //过滤不需要返回
-    typeOf[T] match{
-      case t if t =:= typeOf[Nothing] =>
-      case _ => CallbackHelper.pushCallbackToPool(traceId,callbackDelegate)
-    }
+
+    val callbackPromise = Promise[StreamResponse]
+    CallbackHelper.pushCallbackToPool(traceId,callbackPromise)
     val future = asyncCall(PadPlusServerGrpc.getRequestMethod,request.build())
     future.flatMap{rep=>
       if(rep.getResult != "success"){
-        p.failure(new IllegalAccessException("fail to request ,grpc result:"+rep))
+        callbackPromise.failure(new IllegalAccessException("fail to request ,grpc result:"+rep))
+      }
+      callbackPromise.future
+    }.map { streamResponse =>
+      typeOf[T] match {
+        case t if t =:= typeOf[JsonNode] =>
+          Puppet.objectMapper.readTree(streamResponse.getData).asInstanceOf[T]
+        case _ =>
+          try {
+            Puppet.objectMapper.readValue(streamResponse.getData, classTag.runtimeClass).asInstanceOf[T]
+          } catch {
+            case e: Throwable =>
+              logger.error(e.getMessage, e)
+              throw e
+          }
       }
-      p.future
     }
-//    val response = grpcClient.request(request.build())
-//    logger.debug(s"request $apiType response $response")
-//
-//    if(response.getResult != "success"){
-//      //fail?
-//      logger.error("fail to request grpc,response {}",response)
-//      p.failure(new IllegalAccessException("fail to request ,grpc result:"+response))
-//    }
-//
-//    p.future
   }
+
   type ClientCallback[RespT,T]=RespT => T
   protected def asyncCall[ReqT,RespT](call: MethodDescriptor[ReqT, RespT], req: ReqT): Future[RespT]= {
     asyncCallback(call,req)(resp => resp)

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

@@ -1,6 +1,9 @@
 package wechaty.padplus.support
 
+import java.io.File
+
 import com.google.protobuf.ByteString
+import org.apache.commons.io.FileUtils
 import wechaty.padplus.PuppetPadplus
 import wechaty.padplus.internal.LocalStore
 import wechaty.padplus.schemas.ModelContact.PadplusContactPayload
@@ -69,4 +72,9 @@ trait LocalStoreSupport {
     if(store != null)
       store.close()
   }
+  private[wechaty] def resetLocalStore(): Unit ={
+    stopLocalStore()
+    FileUtils.deleteDirectory(new File(storePath))
+    startLocalStore()
+  }
 }

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

@@ -8,6 +8,8 @@ import wechaty.puppet.schemas.Room
 import wechaty.puppet.schemas.Room.RoomMemberPayload
 import wechaty.puppet.support.RoomMemberSupport
 
+import scala.util.Try
+
 /**
   *
   * @author <a href="mailto:[email protected]">Jun Tsai</a>
@@ -63,16 +65,19 @@ trait RoomMemberRawSupport {
   }
   protected def roomMemberPartialFunction(response:StreamResponse):PartialFunction[ResponseType,Unit]={
     case ResponseType.ROOM_MEMBER_LIST =>
-      val roomMemberList= objectMapper.readValue(response.getData,classOf[GrpcRoomMemberList])
-      val roomId = roomMemberList.roomId
-      val membersStr = roomMemberList.membersJson
-      val membersList =objectMapper.readValue(membersStr,classOf[Array[GrpcRoomMemberPayload]])
-      val data=membersList.map(x=> x.UserName-> convertToPadplusRoomMemberPayload(x)).toMap
-      val padplusRoomMemberMap = new PadplusRoomMemberMap
-      padplusRoomMemberMap.members = data
-      savePadplusRoomMembers(roomId,padplusRoomMemberMap)
+      val roomMemberList       = objectMapper.readValue(response.getData, classOf[GrpcRoomMemberList])
+      val roomId               = roomMemberList.roomId
+      val roomMembers = Try {
+        val membersStr           = roomMemberList.membersJson
+        val membersList          = objectMapper.readValue(membersStr, classOf[Array[GrpcRoomMemberPayload]])
+        val data                 = membersList.map(x => x.UserName -> convertToPadplusRoomMemberPayload(x)).toMap
+        val padplusRoomMemberMap = new PadplusRoomMemberMap
+        padplusRoomMemberMap.members = data
+        savePadplusRoomMembers(roomId, padplusRoomMemberMap)
+        padplusRoomMemberMap
+      }
 
 
-      CallbackHelper.resolveRoomMemberCallback(roomId,padplusRoomMemberMap)
+      CallbackHelper.resolveRoomMemberCallback(roomId,roomMembers)
   }
 }