KERNEL-7725 feat: websocket提供全局拦截器

fine-socketio/src/main/java/com/fr/third/socketio/Configuration.java

@@ -17,11 +17,14 @@ package com.fr.third.socketio;
 
 import java.io.InputStream;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 
 import com.fr.third.socketio.handler.SuccessAuthorizationListener;
 import com.fr.third.socketio.listener.DefaultExceptionListener;
 import com.fr.third.socketio.listener.ExceptionListener;
+import com.fr.third.socketio.listener.interceptor.EventInterceptor;
+import com.fr.third.socketio.listener.interceptor.Interceptor;
 import com.fr.third.socketio.protocol.JsonSupport;
 import com.fr.third.socketio.store.MemoryStoreFactory;
 import com.fr.third.socketio.store.RedissonStoreFactory;
@@ -91,6 +94,10 @@ public class Configuration {
 
     private String accessControlAllowHeaders = null;
 
+    private List<Interceptor> connectInterceptors = Collections.emptyList();
+    private List<Interceptor> disconnectInterceptors = Collections.emptyList();
+    private List<EventInterceptor> eventInterceptors = Collections.emptyList();
+
     public Configuration() {
     }
 
@@ -159,6 +166,9 @@ public class Configuration {
         setRandomSession(conf.randomSession);
 
         setAccessControlAllowHeaders(conf.getAccessControlAllowHeaders());
+        setConnectInterceptors(conf.getConnectInterceptors().toArray(new Interceptor[0]));
+        setDisconnectInterceptors(conf.getDisconnectInterceptors().toArray(new Interceptor[0]));
+        setEventInterceptors(conf.getEventInterceptors().toArray(new EventInterceptor[0]));
     }
 
     public JsonSupport getJsonSupport() {
@@ -596,4 +606,28 @@ public class Configuration {
     public void setAccessControlAllowHeaders(String accessControlAllowHeaders) {
         this.accessControlAllowHeaders = accessControlAllowHeaders;
     }
+
+    public List<Interceptor> getConnectInterceptors() {
+        return connectInterceptors;
+    }
+
+    public void setConnectInterceptors(Interceptor[] connectInterceptors) {
+        this.connectInterceptors = Arrays.asList(connectInterceptors);
+    }
+
+    public List<Interceptor> getDisconnectInterceptors() {
+        return disconnectInterceptors;
+    }
+
+    public void setDisconnectInterceptors(Interceptor[] disconnectInterceptors) {
+        this.disconnectInterceptors = Arrays.asList(disconnectInterceptors);
+    }
+
+    public List<EventInterceptor> getEventInterceptors() {
+        return eventInterceptors;
+    }
+
+    public void setEventInterceptors(EventInterceptor[] eventInterceptors) {
+        this.eventInterceptors = Arrays.asList(eventInterceptors);
+    }
 }

fine-socketio/src/main/java/com/fr/third/socketio/listener/interceptor/EventInterceptor.java

@@ -0,0 +1,27 @@
+package com.fr.third.socketio.listener.interceptor;
+
+import com.fr.third.socketio.AckRequest;
+import com.fr.third.socketio.SocketIOClient;
+
+import java.util.List;
+
+/**
+ * Desc
+ *
+ * @author Elijah
+ * created on 2021-04-26
+ */
+public interface EventInterceptor extends Interceptor {
+    default void before(SocketIOClient client, List<Object> params, AckRequest ackRequest){
+        before(client);
+    }
+
+    default void after(SocketIOClient client, List<Object> params, AckRequest ackRequest){
+        after(client);
+    }
+
+    default boolean accept(String eventName, SocketIOClient client, List<Object> param){
+        return accept(client);
+    }
+
+}

fine-socketio/src/main/java/com/fr/third/socketio/listener/interceptor/Interceptor.java

@@ -0,0 +1,23 @@
+package com.fr.third.socketio.listener.interceptor;
+
+import com.fr.third.socketio.SocketIOClient;
+
+/**
+ * Desc
+ *
+ * @author Elijah
+ * created on 2021-05-06
+ */
+public interface Interceptor {
+    default void before(SocketIOClient client){
+
+    }
+
+    default void after(SocketIOClient client){
+
+    }
+
+    default boolean accept(SocketIOClient client){
+        return true;
+    }
+}

fine-socketio/src/main/java/com/fr/third/socketio/namespace/Namespace.java

@@ -34,6 +34,8 @@ import com.fr.third.socketio.MultiTypeArgs;
 import com.fr.third.socketio.SocketIOClient;
 import com.fr.third.socketio.SocketIONamespace;
 import com.fr.third.socketio.annotation.ScannerEngine;
+import com.fr.third.socketio.listener.interceptor.EventInterceptor;
+import com.fr.third.socketio.listener.interceptor.Interceptor;
 import com.fr.third.socketio.protocol.JsonSupport;
 import com.fr.third.socketio.protocol.Packet;
 import com.fr.third.socketio.store.StoreFactory;
@@ -74,6 +76,7 @@ public class Namespace implements SocketIONamespace {
     private final JsonSupport jsonSupport;
     private final StoreFactory storeFactory;
     private final ExceptionListener exceptionListener;
+    private final Configuration configuration;
 
     public Namespace(String name, Configuration configuration) {
         super();
@@ -82,6 +85,7 @@ public class Namespace implements SocketIONamespace {
         this.storeFactory = configuration.getStoreFactory();
         this.exceptionListener = configuration.getExceptionListener();
         this.ackMode = configuration.getAckMode();
+        this.configuration = configuration;
     }
 
     public void addClient(SocketIOClient client) {
@@ -137,7 +141,16 @@ public class Namespace implements SocketIONamespace {
         if (entry == null) {
             return;
         }
-
+        for (EventInterceptor eventInterceptor: configuration.getEventInterceptors()) {
+            try {
+                //不影响后续执行
+                if (eventInterceptor.accept(eventName, client, args)) {
+                    eventInterceptor.before(client, args, ackRequest);
+                }
+            } catch (Exception e) {
+                exceptionListener.onEventException(e, args, client);
+            }
+        }
         try {
             Queue<DataListener> listeners = entry.getListeners();
             for (DataListener dataListener : listeners) {
@@ -149,6 +162,17 @@ public class Namespace implements SocketIONamespace {
             if (ackMode == AckMode.AUTO_SUCCESS_ONLY) {
                 return;
             }
+        } finally {
+            for (EventInterceptor eventInterceptor: configuration.getEventInterceptors()) {
+                try {
+                    //不影响后续执行
+                    if (eventInterceptor.accept(eventName, client, args)) {
+                        eventInterceptor.after(client, args, ackRequest);
+                    }
+                } catch (Exception e) {
+                    exceptionListener.onEventException(e, args, client);
+                }
+            }
         }
 
         sendAck(ackRequest);
@@ -190,12 +214,15 @@ public class Namespace implements SocketIONamespace {
         }
         clientRooms.remove(client.getSessionId());
 
+        callInterceptor(client, false, true);
         try {
             for (DisconnectListener listener : disconnectListeners) {
                 listener.onDisconnect(client);
             }
         } catch (Exception e) {
             exceptionListener.onDisconnectException(e, client);
+        } finally {
+            callInterceptor(client, false, false);
         }
     }
 
@@ -208,15 +235,39 @@ public class Namespace implements SocketIONamespace {
         join(getName(), client.getSessionId());
         storeFactory.pubSubStore().publish(PubSubType.JOIN, new JoinLeaveMessage(client.getSessionId(), getName(), getName()));
 
+        callInterceptor(client, true, true);
         try {
             for (ConnectListener listener : connectListeners) {
                 listener.onConnect(client);
             }
         } catch (Exception e) {
             exceptionListener.onConnectException(e, client);
+        } finally {
+            callInterceptor(client, true, false);
         }
     }
 
+    private void callInterceptor(SocketIOClient client, boolean connect, boolean before) {
+        List<Interceptor> interceptors = connect ? configuration.getConnectInterceptors() : configuration.getDisconnectInterceptors();
+        for (Interceptor interceptor: interceptors) {
+            try{
+                if (interceptor.accept(client)) {
+                    if (before) {
+                        interceptor.before(client);
+                    } else {
+                        interceptor.after(client);
+                    }
+                }
+            } catch (Exception e) {
+                if (connect) {
+                    exceptionListener.onConnectException(e, client);
+                } else {
+                    exceptionListener.onDisconnectException(e, client);
+                }
+            }
+
+        }
+    }
     @Override
     public void addPingListener(PingListener listener) {
         pingListeners.add(listener);