rpc: introduce OncRpcSvcBuilder#withCallInterceptor

Motivation:
Sometimes code need to perform an action before RPC request is executed,
for example:

- logging
- validation
- DoS protection / rate limitation

```
  limiter = RateLimiter.create(500);

  svc = new OncRpcSvcBuilder()
        .withServiceName("svc")
        .withCallInterceptor(c -> limiter.acquire())
        .build();
  svc.start();
```

Modification:
introduce OncRpcSvcBuilder#withCallInterceptor that injects a RPC call
consumer that is called before real work is done.

Result:
additional actions can be performed before rpc call execution.

Acked-by: Lea Morschel
Acked-by: Paul Millar
Target: master

Author: kofemann

oncrpc4j-core/src/main/java/org/dcache/oncrpc4j/rpc/OncRpcSvc.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2009 - 2021 Deutsches Elektronen-Synchroton,
+ * Copyright (c) 2009 - 2022 Deutsches Elektronen-Synchroton,
  * Member of the Helmholtz Association, (DESY), HAMBURG, GERMANY
  *
  * This library is free software; you can redistribute it and/or modify
@@ -67,6 +67,7 @@
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.function.Consumer;
 import javax.net.ssl.SSLContext;
 
 import static com.google.common.base.Throwables.getRootCause;
@@ -129,6 +130,11 @@ public class OncRpcSvc {
      */
     private final String _svcName;
 
+    /**
+     *  {@code java.util.function.Consumer} that is called before RPC request executed.
+     */
+    private final Consumer<RpcCall> _callInterceptor;
+
     /**
      * Create new RPC service with defined configuration.
      * @param builder to build this service
@@ -192,6 +198,7 @@ public class OncRpcSvc {
         _sslContext = builder.getSSLContext();
         _startTLS = builder.isStartTLS();
         _sslParams = builder.getSSLParameters();
+        _callInterceptor = builder.getCallInterceptor();
     }
 
     /**
@@ -346,7 +353,7 @@ public void start() throws IOException {
             if (_gssSessionManager != null) {
                 filterChain.add(new GssProtocolFilter(_gssSessionManager));
             }
-            filterChain.add(new RpcDispatcher(_requestExecutor, _programs, _withSubjectPropagation));
+            filterChain.add(new RpcDispatcher(_requestExecutor, _programs, _withSubjectPropagation, _callInterceptor));
 
             final FilterChain filters = filterChain.build();
 

oncrpc4j-core/src/main/java/org/dcache/oncrpc4j/rpc/OncRpcSvcBuilder.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2009 - 2019 Deutsches Elektronen-Synchroton,
+ * Copyright (c) 2009 - 2022 Deutsches Elektronen-Synchroton,
  * Member of the Helmholtz Association, (DESY), HAMBURG, GERMANY
  *
  * This library is free software; you can redistribute it and/or modify
@@ -29,6 +29,7 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ThreadFactory;
+import java.util.function.Consumer;
 import javax.net.ssl.SSLContext;
 import javax.net.ssl.SSLParameters;
 
@@ -84,6 +85,8 @@ public class OncRpcSvcBuilder {
     private SSLParameters _sslParams;
     private MemoryAllocator _allocator = MemoryAllocator.DEFAULT;
 
+    private Consumer<RpcCall> _callInterceptor = c -> {};
+
     public OncRpcSvcBuilder withAutoPublish() {
         _autoPublish = true;
         return this;
@@ -282,6 +285,15 @@ public GssSessionManager getGssSessionManager() {
         return _gssSessionManager;
     }
 
+    public OncRpcSvcBuilder withCallInterceptor(Consumer<RpcCall> interceptor) {
+        _callInterceptor = interceptor;
+        return this;
+    }
+
+    public Consumer<RpcCall> getCallInterceptor() {
+        return _callInterceptor;
+    }
+
     public ExecutorService getWorkerThreadExecutorService() {
         if (_ioStrategy == IoStrategy.SAME_THREAD ) {
             return MoreExecutors.newDirectExecutorService();

oncrpc4j-core/src/main/java/org/dcache/oncrpc4j/rpc/RpcDispatcher.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2009 - 2018 Deutsches Elektronen-Synchroton,
+ * Copyright (c) 2009 - 2022 Deutsches Elektronen-Synchroton,
  * Member of the Helmholtz Association, (DESY), HAMBURG, GERMANY
  *
  * This library is free software; you can redistribute it and/or modify
@@ -25,6 +25,7 @@
 import java.security.PrivilegedExceptionAction;
 import java.util.Map;
 import java.util.concurrent.ExecutorService;
+import java.util.function.Consumer;
 
 import com.google.common.base.Throwables;
 import org.slf4j.Logger;
@@ -54,6 +55,12 @@ public class RpcDispatcher extends BaseFilter {
      * from request credentials.
      */
     private final boolean _withSubjectPropagation;
+
+    /**
+     *  {@code java.util.function.Consumer} that is called before RPC request executed.
+     */
+    private final Consumer<RpcCall> _callInterceptor;
+
     /**
      * Create new RPC dispatcher for given program.
      *
@@ -62,16 +69,18 @@ public class RpcDispatcher extends BaseFilter {
      *     with a mapping between program number and program
      *     handler.
      * @param withSubjectPropagation use {@link Subject#doAs} to exacerbate request.
+     * @param callInterceptor consumer that will be called before the dispatcher performs its real work.
      *
      * @throws NullPointerException if executor or program is null
      */
     public RpcDispatcher(ExecutorService executor, Map<OncRpcProgram,
-            RpcDispatchable> programs, boolean withSubjectPropagation)
+            RpcDispatchable> programs, boolean withSubjectPropagation, Consumer<RpcCall> callInterceptor)
             throws NullPointerException {
 
         _programs = requireNonNull(programs, "Programs is NULL");
         _asyncExecutorService = requireNonNull(executor, "ExecutorService is NULL");
         _withSubjectPropagation = withSubjectPropagation;
+        _callInterceptor = callInterceptor;
     }
 
     @Override
@@ -91,6 +100,9 @@ public NextAction handleRead(final FilterChainContext ctx) throws IOException {
             _asyncExecutorService.execute(new Runnable() {
                 @Override
                 public void run() {
+
+                    _callInterceptor.accept(call);
+
                     try {
                         if (_withSubjectPropagation) {
                             Subject subject = call.getCredential().getSubject();