Add DisrutporSupport,use new thread to process event

Author: jcai

pom.xml

@@ -131,6 +131,11 @@
         <artifactId>aws-java-sdk-s3</artifactId>
         <version>1.11.812</version>
       </dependency>
+      <dependency>
+        <groupId>com.lmax</groupId>
+        <artifactId>disruptor</artifactId>
+        <version>3.4.2</version>
+      </dependency>
       <dependency>
         <groupId>org.grpcmock</groupId>
         <artifactId>grpcmock-junit5</artifactId>

wechaty-puppet-padplus/pom.xml

@@ -12,6 +12,10 @@
 
 
   <dependencies>
+    <dependency>
+      <groupId>com.lmax</groupId>
+      <artifactId>disruptor</artifactId>
+    </dependency>
     <dependency>
       <groupId>com.amazonaws</groupId>
       <artifactId>aws-java-sdk-s3</artifactId>

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

@@ -24,9 +24,11 @@ class PuppetPadplus(val option:PuppetOptions,val storePath:String="/tmp/padplus"
     with GrpcSupport
     with GrpcEventSupport
     with LocalStoreSupport
+    with DisrutporSupport
     with LazyLogging {
   protected var uinOpt:Option[String]=None
   def start(): Unit ={
+    startDisruptor()
     startGrpc(option.endPoint.get)
     //waiting stream start....
     logger.info("waiting stream start....")
@@ -42,12 +44,15 @@ class PuppetPadplus(val option:PuppetOptions,val storePath:String="/tmp/padplus"
     }
   }
   def stop(): Unit = {
+    shutdownDisruptor()
     stopGrpc()
     stopLocalStore()
   }
 
   override def selfIdOpt(): Option[String] = selfId
 
+
+
   /**
     *
     * Friendship

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

@@ -0,0 +1,68 @@
+package wechaty.padplus.support
+
+import java.util.concurrent.{Executors, TimeUnit}
+
+import com.lmax.disruptor.dsl.{Disruptor, ProducerType}
+import com.lmax.disruptor.{BusySpinWaitStrategy, EventFactory, EventHandler, ExceptionHandler}
+import com.typesafe.scalalogging.LazyLogging
+import wechaty.puppet.events.EventEmitter
+import wechaty.puppet.schemas.Puppet.PuppetEventName
+import wechaty.puppet.schemas.Puppet.PuppetEventName.Type
+
+/**
+  *
+  * @author <a href="mailto:[email protected]">Jun Tsai</a>
+  * @since 2020-06-30
+  */
+trait DisrutporSupport extends EventEmitter {
+  self :LazyLogging =>
+  private class EventDef {
+    var eventName: PuppetEventName.Type = _
+    var value    : Any = _
+  }
+  private var disruptor:Disruptor[EventDef] = _
+  def startDisruptor(): Unit ={
+    disruptor = new Disruptor[EventDef](
+      new EventFactory[EventDef]() {
+        @Override
+        def newInstance():EventDef= {
+          new EventDef()
+        }
+      },
+      1 << 32,
+      Executors.defaultThreadFactory(),
+      ProducerType.SINGLE,
+      new BusySpinWaitStrategy()
+    )
+    disruptor.handleEventsWith(new EventHandler[EventDef](){
+      override def onEvent(event: EventDef, sequence: Long, endOfBatch: Boolean): Unit = {
+        logger.debug("emit event using super {}",event.eventName)
+        DisrutporSupport.super.emit(event.eventName,event.value)
+      }
+    })
+    disruptor.setDefaultExceptionHandler(new ExceptionHandler[EventDef] {
+      override def handleEventException(ex: Throwable, sequence: Long, event: EventDef): Unit = {
+        logger.error(ex.getMessage,ex)
+      }
+
+      override def handleOnStartException(ex: Throwable): Unit = {
+        throw ex
+      }
+
+      override def handleOnShutdownException(ex: Throwable): Unit = {
+        logger.error(ex.getMessage,ex)
+      }
+    })
+    disruptor.start()
+  }
+  override def emit[T](eventName: Type, value: T): Unit = {
+    logger.debug("publish event {}",eventName)
+    disruptor.publishEvent((event: EventDef, sequence: Long) => {
+      event.eventName = eventName
+      event.value = value
+    })
+  }
+  protected def shutdownDisruptor(): Unit ={
+    disruptor.shutdown(5,TimeUnit.SECONDS);
+  }
+}

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

@@ -98,16 +98,6 @@ trait GrpcSupport {
     logger.info("start grpc client ....")
     this.grpcClient = PadPlusServerGrpc.newBlockingStub(channel)
     startStream()
-
-//    this.grpcClient.start(Base.StartRequest.newBuilder().build())
-//
-//    try{
-//      //sometime the grpc can't work well,so logout before start bot
-//      //      this.grpcClient.logout(Base.LogoutRequest.newBuilder().build())
-//    }catch{
-//      case e:Throwable=>
-//        logger.warn(e.getMessage)
-//    }
     logger.info("start grpc client done")
   }
 

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

@@ -68,9 +68,8 @@ trait MessageRawSupport {
     json.put("messageType",PadplusMessageType.Text.id)
     json.put("fromUserName",selfId.get)
     json.put("toUserName",conversationId)
-    val payload = asyncRequest[GrpcMessagePayload](ApiType.SEND_MESSAGE,Some(json.toString))
-//    payload.msgId
-    null
+    val payload = syncRequest[GrpcMessagePayload](ApiType.SEND_MESSAGE,Some(json.toString))
+    payload.MsgId
   }
 
   override def messageSendUrl(conversationId: String, urlLinkPayload: UrlLink.UrlLinkPayload): String = ???