From ad2b0fb6c3f3055567756bef24f7f11329b2b9b6 Mon Sep 17 00:00:00 2001
From: "Frank.R.Wu" <wukd94@pku.edu.cn>
Date: Sun, 31 Oct 2021 23:07:14 +0800
Subject: [PATCH] feat: update event mechanism

add event type local and global, clients have to use contractID and topic to subscribe local event; allow clients to subscribe topics (will not be recorded)
---
 contracts/EventPublisher/EventPublisher.yjs   |  4 +-
 .../ws/ContractManagerFrameHandler.java       | 66 +++++++++++--------
 .../org/bdware/server/ws/EventActions.java    | 30 +++++++++
 3 files changed, 69 insertions(+), 31 deletions(-)
 create mode 100644 src/main/java/org/bdware/server/ws/EventActions.java

diff --git a/contracts/EventPublisher/EventPublisher.yjs b/contracts/EventPublisher/EventPublisher.yjs
index b2e66f5..9eb1254 100644
--- a/contracts/EventPublisher/EventPublisher.yjs
+++ b/contracts/EventPublisher/EventPublisher.yjs
@@ -7,7 +7,7 @@ contract EventPublisher {
    * function te: publish an event with topic 'te' and default semantics AT_LEAST_ONCE
    * function tes: publish an event with topic 'te' and another semantics
    */
-  event te;
+  event global te;
   /*
    * event declaration with semantics
    * valid semantics: AT_LEASE_ONCE, AT_MOST_ONCE, and ONLY_ONCE
@@ -15,7 +15,7 @@ contract EventPublisher {
    * function tews: publish an event with topic 'tews' and semantics AT_MOST_ONCE
    * function tewss: publish an event with topic 'tews' and another semantics
    */
-  event AT_MOST_ONCE tews;
+  event global tews(AT_MOST_ONCE);
   // publish an event with declared topic and default semantics AT_LEAST_ONCE
   export function pub1(e) {
     var content = YancloudUtil.random() + ' EventPublisher.pub1';
diff --git a/src/main/java/org/bdware/server/ws/ContractManagerFrameHandler.java b/src/main/java/org/bdware/server/ws/ContractManagerFrameHandler.java
index ae8e166..78a46c8 100644
--- a/src/main/java/org/bdware/server/ws/ContractManagerFrameHandler.java
+++ b/src/main/java/org/bdware/server/ws/ContractManagerFrameHandler.java
@@ -56,7 +56,9 @@ public class ContractManagerFrameHandler extends SimpleChannelInboundHandler<Web
                         userManagerAction,
                         new MasterWSAction(userManagerAction), // 多节点执行
                         new CMLogAction(),
-                        new ProcessAction(), GRPCPool.instance) {
+                        new ProcessAction(),
+                        GRPCPool.instance,
+                        new EventActions()) {
                     @Override
                     public boolean checkPermission(Action a, JsonObject arg, long permission) {
                         long val = a.userPermission();
@@ -74,25 +76,21 @@ public class ContractManagerFrameHandler extends SimpleChannelInboundHandler<Web
                             flag = false;
                         }
                         LOGGER.debug(
-                                "[checkpermission] "
-                                        + action
-                                        + " val:"
-                                        + val
-                                        + " permission:"
-                                        + permission
-                                        + " status:"
-                                        + status);
+                                String.format(
+                                        "%s val:%d permission:%d status:%s",
+                                        action,
+                                        val,
+                                        permission,
+                                        status));
 
-                        String sb = "{\"action\":\"" +
-                                action +
-                                "\",\"pubKey\":\"" +
-                                userManagerAction.getPubKey() +
-                                "\",\"status\":\"" +
-                                status +
-                                "\",\"date\":" +
-                                System.currentTimeMillis() +
-                                "}";
-                        CMHttpServer.nodeLogDB.put(action, sb);
+                        CMHttpServer.nodeLogDB.put(
+                                action,
+                                String.format(
+                                        "{\"action\":\"%s\",\"pubKey\":\"%s\",\"status\":\"%s\",\"date\":%d}",
+                                        action,
+                                        userManagerAction.getPubKey(),
+                                        status,
+                                        System.currentTimeMillis()));
                         // TimeDBUtil.instance.put(CMTables.LocalNodeLogDB.toString(),
                         // sb.toString());
                         return flag;
@@ -113,8 +111,7 @@ public class ContractManagerFrameHandler extends SimpleChannelInboundHandler<Web
     }
 
     @Override
-    protected void channelRead0(final ChannelHandlerContext ctx, WebSocketFrame frame)
-            throws Exception {
+    protected void channelRead0(final ChannelHandlerContext ctx, WebSocketFrame frame) {
         // ping and pong frames already handled
         if (frame instanceof TextWebSocketFrame) {
             // Send the uppercase string back.
@@ -166,7 +163,7 @@ public class ContractManagerFrameHandler extends SimpleChannelInboundHandler<Web
                 ae.handle(
                         action,
                         map,
-                        new ResultCallback() {
+                        new ResultCallback(ctx.channel()) {
                             @Override
                             public void onResult(String ret) {
                                 if (ret != null) {
@@ -181,7 +178,6 @@ public class ContractManagerFrameHandler extends SimpleChannelInboundHandler<Web
                 ctx.channel().writeAndFlush(new TextWebSocketFrame((JsonUtil.toJson(response))));
             } catch (Exception e) {
                 ByteArrayOutputStream bo = new ByteArrayOutputStream();
-                e.printStackTrace();
                 e.printStackTrace(new PrintStream(bo));
                 response = new Response();
                 response.action = "onException";
@@ -189,12 +185,24 @@ public class ContractManagerFrameHandler extends SimpleChannelInboundHandler<Web
                 StringBuilder ret = new StringBuilder();
                 int count = 0;
                 for (String s : strs) {
-                    if (s.contains("sun.reflect")) continue;
-                    if (s.contains("java.lang.reflect")) continue;
-                    if (s.contains("org.apache")) continue;
-                    if (s.contains("java.util")) continue;
-                    if (s.contains("java.lang")) continue;
-                    if (s.contains("io.netty")) continue;
+                    if (s.contains("sun.reflect")) {
+                        continue;
+                    }
+                    if (s.contains("java.lang.reflect")) {
+                        continue;
+                    }
+                    if (s.contains("org.apache")) {
+                        continue;
+                    }
+                    if (s.contains("java.util")) {
+                        continue;
+                    }
+                    if (s.contains("java.lang")) {
+                        continue;
+                    }
+                    if (s.contains("io.netty")) {
+                        continue;
+                    }
                     ret.append(s);
                     ret.append("\n");
                     if (count++ > 5) break;
diff --git a/src/main/java/org/bdware/server/ws/EventActions.java b/src/main/java/org/bdware/server/ws/EventActions.java
new file mode 100644
index 0000000..4b4a2a7
--- /dev/null
+++ b/src/main/java/org/bdware/server/ws/EventActions.java
@@ -0,0 +1,30 @@
+package org.bdware.server.ws;
+
+import com.google.gson.JsonObject;
+import org.bdware.sc.ContractClient;
+import org.bdware.sc.conn.ResultCallback;
+import org.bdware.sc.util.HashUtil;
+import org.bdware.server.action.Action;
+import org.bdware.server.action.CMActions;
+
+public class EventActions {
+    @Action(async = true, userPermission = 0)
+    public void subEvent(JsonObject args, final ResultCallback rcb) {
+        if (!args.has("topic")) {
+            rcb.onResult("{\"status\":\"Error\",\"data\":\"no topic arg!\"}");
+            return;
+        }
+        String topic = args.get("topic").getAsString();
+        if (args.has("contractID")) {
+            String argCID = args.get("contractID").getAsString();
+            ContractClient client = CMActions.manager.getClient(argCID);
+            if (null == client) {
+                rcb.onResult("{\"status\":\"Error\",\"data\":\"invalid contract ID or Name!\"}");
+                return;
+            }
+            String contractID = client.getContractID();
+            topic = HashUtil.sha3(contractID + topic);
+        }
+        CMActions.manager.subEventByClient(topic, rcb.getChannel());
+    }
+}