feature #183: .addEmitFilter and .addCallFilter

nodel-framework/src/main/java/org/nodel/core/NodelServerAction.java

@@ -84,6 +84,11 @@ public Map<String, Object> getFullSchema() {
      * For other in-process 'call' handlers that may be interested in this action.
      */
     private AtomicReference<Handler.H1<Object>[]> _callHandlers = new AtomicReference<Handler.H1<Object>[]>();
+
+    /**
+     * Allow a call filter which can trap and alter the value of the arg
+     */
+    private Handler.F1<Object, Object> _callFilter = null;
 
     private boolean _closed;
 
@@ -257,13 +262,34 @@ protected void handleActionRequest(Object arg) {
 
         @Override
         public void handleActionRequest(Object arg) {
+            if (_callFilter != null) {
+                // arg = _emitFilter.handle(arg);
+                if (_callbackQueue != null) {
+                    try {
+                        arg = _callbackQueue.handle(_callFilter, arg);
+                    } catch (Exception e) {
+                        // handle gracefully ...
+                        _exceptionHandler.handle(e);
+                        // ... and keep going with arg regardless
+                    }
+                } else {
+                    try {
+                        arg = _callFilter.handle(arg);
+                    } catch (Exception exc) {
+                        throw new RuntimeException("Emit filter", exc);
+                    }
+                }
+            }
+
+            final Object finalArg = arg;
+
             _argValue.set(arg);
             _timestamp.set(DateTime.now());
 
             // seq must be set last
             _seqNum = Nodel.getNextSeq();
 
-            _handler.handleActionRequest(arg);
+            _handler.handleActionRequest(finalArg);
 
             // snap-shot of handlers
             final H1<Object>[] handlers = _callHandlers.get();
@@ -283,10 +309,10 @@ public void run() {
                         // call handlers one after the other
                         for (Handler.H1<Object> handler : handlers) {
                             if (_callbackQueue != null)
-                                _callbackQueue.handle(handler, arg, _exceptionHandler);
+                                _callbackQueue.handle(handler, finalArg, _exceptionHandler);
                             else {
                                 try {
-                                    Handler.tryHandle(handler, arg);
+                                    Handler.tryHandle(handler, finalArg);
                                 } catch (Exception exc) {
                                     lastExc = exc;
                                 }
@@ -350,6 +376,13 @@ public void addCallHandler(Handler.H1<Object> handler) {
 
             // otherwise keep trying
         }
+    }
+
+    /**
+     * Filtering the arg emitted 
+     */
+    public void addCallFilter(Handler.F1<Object, Object> filter) {
+        _callFilter = filter;
     }    
 
     /**

nodel-framework/src/main/java/org/nodel/core/NodelServerEvent.java

@@ -55,6 +55,11 @@ public class NodelServerEvent implements Closeable {
      * For other in-process 'emit' handlers that may be interested in this event.
      */
     private LockFreeList<Handler.H1<Object>> _emitHandlers = new LockFreeList<>();
+
+    /**
+     * Allow an emit filter which can trap and alter the value of the arg
+     */
+    private Handler.F1<Object, Object> _emitFilter = null;
 
     private String _title;
 
@@ -300,11 +305,30 @@ public void emit(@Param(name="arg", title="Argument") Object arg) {
     /**
      * Fires the event.
      */
-    private void doEmit(final Object arg) {
+    private void doEmit(Object arg) {
         DateTime now = DateTime.now();
 
+        if (_emitFilter != null) {
+            // arg = _emitFilter.handle(arg);
+            if (_callbackQueue != null) {
+                try {
+                    arg = _callbackQueue.handle(_emitFilter, arg);
+                } catch (Exception e) {
+                    // handle gracefully ...
+                    _exceptionHandler.handle(e);
+                    // ... and keep going with arg regardless
+                }
+            } else {
+                try {
+                    arg = _emitFilter.handle(arg);
+                } catch (Exception exc) {
+                    throw new RuntimeException("Emit filter", exc);
+                }
+            }
+        }
+
         ArgInstance argInstance = new ArgInstance();
-        argInstance.timestamp = DateTime.now();
+        argInstance.timestamp = now;
         argInstance.arg = arg;
         argInstance.seqNum = Nodel.getNextSeq(); 
 
@@ -318,6 +342,8 @@ private void doEmit(final Object arg) {
         // snap-shot of handlers
         final List<Handler.H1<Object>> handlers = _emitHandlers.items();
 
+        final Object finalArg = arg;
+
         // if there are some handlers, use the Channel Client thread-pool (treat as though remote events)
         if (handlers.size() > 0) {
             ChannelClient.getThreadPool().execute(new Runnable() {
@@ -333,10 +359,10 @@ public void run() {
                     // call handlers one after the other
                     for (Handler.H1<Object> handler : handlers) {
                         if (_callbackQueue != null)
-                            _callbackQueue.handle(handler, arg, _exceptionHandler);
+                            _callbackQueue.handle(handler, finalArg, _exceptionHandler);
                         else {
                             try {
-                                Handler.handle(handler, arg);
+                                Handler.handle(handler, finalArg);
                             } catch (Exception exc) {
                                 lastExc = exc;
                             }
@@ -359,6 +385,13 @@ public void run() {
     public void addEmitHandler(Handler.H1<Object> handler) {
         _emitHandlers.add(handler);
     }
+
+    /**
+     * Filtering the arg emitted 
+     */
+    public void addEmitFilter(Handler.F1<Object, Object> filter) {
+        _emitFilter = filter;
+    }
 
     /**
      * Attaches a monitor.

nodel-framework/src/main/java/org/nodel/threading/CallbackQueue.java

@@ -55,6 +55,20 @@ public <T> T handle(Callable<T> func) throws Exception {
             _fairLock.unlock();
         }
     }
+
+    /**
+     * For synchronous functions.
+     */
+    public <R, T> R handle(Handler.F1<R, T> func, T arg) throws Exception {
+        try {
+            _fairLock.lock();
+
+            return func.handle(arg);
+
+        } finally {
+            _fairLock.unlock();
+        }
+    }    
 
     /**
      * Creates a callback instance.