refactoring

Author: jlangch

src/main/java/com/github/jlangch/venice/impl/functions/IPCFunctions.java

@@ -102,6 +102,11 @@ public class IPCFunctions {
                                                         " The cutoff size can be specified as a number like `1000`" +
                                                         " or a number with a unit like `:1KB` or `:2MB`.¶" +
                                                         " Defaults to -1 (no compression)|\n" +
+                        "| :encrypt b                   | If `true` encrypt the payload data of all messages exchanged" +
+                                                        " with this server.¶" +
+                                                        " The data is AES-256-GCM encrypted using a secret that is" +
+                                                        " created and exchanged using the Diffie-Hellman key exchange " +
+                                                        " algorithm.|\n\n" +
                         "| :write-ahead-log-dir f       | Provide a write-ahead-log directory to support durable queues.¶" +
                                                         " Defaults to `nil`.|\n" +
                         "| :write-ahead-log-compress b  | If `true` compresses the write-ahead-log records.¶" +
@@ -164,6 +169,7 @@ public VncVal apply(final VncList args) {
                 final VncVal maxMsgSizeVal = options.get(new VncKeyword("max-message-size"));
                 final VncVal maxMaxQueuesVal = options.get(new VncKeyword("max-queues"));
                 final VncVal compressCutoffSizeVal = options.get(new VncKeyword("compress-cutoff-size"));
+                final VncVal encryptVal = options.get(new VncKeyword("encrypt"), VncBoolean.False);
                 final VncVal walDirVal = options.get(new VncKeyword("write-ahead-log-dir"));
                 final VncVal walCompressVal = options.get(new VncKeyword("write-ahead-log-compress"));
                 final VncVal walCompactAtStartVal = options.get(new VncKeyword("write-ahead-log-compact"));
@@ -175,6 +181,7 @@ public VncVal apply(final VncList args) {
                 final long maxMsgSize = convertMaxMessageSizeToLong(maxMsgSizeVal);
                 final long maxQueues = convertMaxMessageSizeToLong(maxMaxQueuesVal);
                 final long compressCutoffSize = convertMaxMessageSizeToLong(compressCutoffSizeVal);
+                final boolean encrypt = Coerce.toVncBoolean(encryptVal).getValue();
 
                 final File walDir = walDirVal == Nil
                                     ? null
@@ -233,6 +240,8 @@ public VncVal apply(final VncList args) {
                     server.setCompressCutoffSize(compressCutoffSize);
                 }
 
+                server.setEncryption(encrypt);
+
                 if (walDir != null) {
                     server.enableWriteAheadLog(walDir, walCompress, walCompactAtStart);
                 }
@@ -268,10 +277,6 @@ public VncVal apply(final VncList args) {
                         "| port p | The server's TCP/IP port |\n" +
                         "| host h | The server's TCP/IP host |\n\n" +
                         "*Options:* \n\n" +
-                        "| :max-message-size n     | The max size of the message payload." +
-                                                   " Defaults to `200MB`.¶" +
-                                                   " The max size can be specified as a number like `20000`" +
-                                                   " or a number with a unit like `:20KB` or `:20MB`|\n" +
                         "| :compress-cutoff-size n | The compression cutoff size for payload messages.¶" +
                                                    " With a negative cutoff size payload messages will not be" +
                                                    " compressed. If the payload message size is greater than the" +
@@ -353,19 +358,15 @@ else if ( args.size() == 2) {
                     final int port = Coerce.toVncLong(args.second()).getIntValue();
 
                     final VncHashMap options = VncHashMap.ofAll(args.slice(2));
-                    final VncVal maxMsgSizeVal = options.get(new VncKeyword("max-message-size"));
                     final VncVal compressCutoffSizeVal = options.get(new VncKeyword("compress-cutoff-size"));
                     final VncVal encryptVal = options.get(new VncKeyword("encrypt"), VncBoolean.False);
 
-                    final long maxMsgSize = convertMaxMessageSizeToLong(maxMsgSizeVal);
                     final long compressCutoffSize = convertMaxMessageSizeToLong(compressCutoffSizeVal);
                     final boolean encrypt = Coerce.toVncBoolean(encryptVal).getValue();
 
-                    final TcpClient client = new TcpClient(host, port, encrypt);
+                    final TcpClient client = new TcpClient(host, port);
 
-                    if (maxMsgSize > 0) {
-                        client.setMaximumMessageSize(maxMsgSize);
-                    }
+                    client.setEncryption(encrypt);
 
                     if (compressCutoffSize >= 0) {
                         client.setCompressCutoffSize(compressCutoffSize);

src/main/java/com/github/jlangch/venice/util/ipc/MessageType.java

@@ -40,7 +40,9 @@ public enum MessageType {
     REMOVE_QUEUE(32),
     STATUS_QUEUE(33),
 
-    DIFFIE_HELLMAN_KEY_REQUEST(40),
+    CLIENT_CONFIG(40),
+
+    DIFFIE_HELLMAN_KEY_REQUEST(50),
 
 
     // responses

src/main/java/com/github/jlangch/venice/util/ipc/TcpClient.java

@@ -46,9 +46,11 @@
 import com.github.jlangch.venice.impl.threadpool.ManagedCachedThreadPoolExecutor;
 import com.github.jlangch.venice.impl.types.VncBoolean;
 import com.github.jlangch.venice.impl.types.VncKeyword;
+import com.github.jlangch.venice.impl.types.VncLong;
 import com.github.jlangch.venice.impl.types.VncVal;
 import com.github.jlangch.venice.impl.types.collections.VncMap;
 import com.github.jlangch.venice.impl.types.collections.VncOrderedMap;
+import com.github.jlangch.venice.impl.types.util.Coerce;
 import com.github.jlangch.venice.impl.util.CollectionUtil;
 import com.github.jlangch.venice.impl.util.StringUtil;
 import com.github.jlangch.venice.util.dh.DiffieHellmanKeys;
@@ -76,7 +78,7 @@ public class TcpClient implements Cloneable, Closeable {
      * @param port a port
      */
     public TcpClient(final int port) {
-        this(null, port, false);
+        this(null, port);
     }
 
     /**
@@ -91,61 +93,42 @@ public TcpClient(final int port) {
      * @param port a port
      */
     public TcpClient(final String host, final int port) {
-        this(host, port, false);
-    }
-
-    /**
-     * Create a new TcpClient connecting to a TcpServer on the local host
-     * and port
-     *
-     * <p>The client is NOT thread safe!
-     *
-     * <p>The client must be closed after use!
-     *
-     * @param port a port
-     * @param encrypt if <code>true</code> encrypt the payload data at transport
-     *                level communication between this client and the server.
-     */
-    public TcpClient(final int port, final boolean encrypt) {
-        this(null, port, encrypt);
-    }
-
-    /**
-     * Create a new TcpClient connecting to a TcpServer on the specified host
-     * and port
-     *
-     * <p>The client is NOT thread safe!
-     *
-     * <p>The client must be closed after use!
-     *
-     * @param host a host
-     * @param port a port
-     * @param encrypt if <code>true</code> encrypt the payload data at transport
-     *                level communication between this client and the server.
-     */
-    public TcpClient(final String host, final int port, final boolean encrypt) {
         this.host = StringUtil.isBlank(host) ? "127.0.0.1" : host;
         this.port = port;
-        this.encrypt = encrypt;
         this.endpointId = UUID.randomUUID().toString();
         this.dhKeys = DiffieHellmanKeys.create();
     }
 
 
     @Override
     public Object clone() {
-        final TcpClient client = new TcpClient(host, port, isEncrypted());
+        final TcpClient client = new TcpClient(host, port);
+        client.setEncryption(isEncrypted());
         client.setCompressCutoffSize(getCompressCutoffSize());
-        client.setMaximumMessageSize(getMaximumMessageSize());
-        return client;
+         return client;
+    }
+
+    /**
+     * Set the encryption mode
+     *
+     * @param encrypt if <code>true</code> encrypt the payload data at transport
+     *                level communication between this client and the server.
+     */
+    public void setEncryption(final boolean encrypt) {
+        if (opened.get()) {
+            throw new VncException(
+                   "The encryption mode cannot be set anymore "
+                   + "once the client has been opened!");
+        }
+        this.encrypt.set(encrypt);
     }
 
     /**
      * @return <code>true</code> if this client has transport level encryption
      *         enabled else <code>false</code>
      */
     public boolean isEncrypted() {
-        return encryptor.get().isActive();
+        return encrypt.get();
     }
 
     /**
@@ -177,23 +160,15 @@ public long getCompressCutoffSize() {
         return compressor.get().cutoffSize();
     }
 
-    /**
-     * Set the maximum message size.
-     *
-     * <p>Defaults to 200 MB
-     *
-     * @param maxSize the max message size in bytes
-     * @return this client
-     */
-    public TcpClient setMaximumMessageSize(final long maxSize) {
-        maxMessageSize.set(Math.max(MESSAGE_LIMIT_MIN, Math.min(MESSAGE_LIMIT_MAX, maxSize)));
-        return this;
-    }
-
     /**
      * @return return the client's max message size
      */
     public long getMaximumMessageSize() {
+        if (!opened.get()) {
+            throw new VncException(
+                   "The max message size cannot be queried if the client has "
+                   + "been opened and has requested the size from the server!");
+        }
         return maxMessageSize.get();
     }
 
@@ -237,7 +212,37 @@ public void open() {
                         ex);
             }
 
-            if (encrypt) {
+            // request config from server
+            try {
+                final IMessage response = send(createConfigRequestMessage());
+                if (response.getResponseStatus() != ResponseStatus.OK) {
+                    throw new VncException(
+                            "Failed to get client config from server. Server answered with "
+                            + response.getResponseStatus() + " !");
+                }
+
+                // handle config
+                final VncMap config = (VncMap)((Message)response).getVeniceData();
+                maxMessageSize.set(
+                    Coerce.toVncLong(
+                        config.get(
+                            new VncKeyword("max-msg-size"),
+                            new VncLong(MESSAGE_LIMIT_MAX))).toJavaLong());
+                encrypt.set(
+                    encrypt.get()             // client side encrypt request
+                    || VncBoolean.isTrue(     // server side encrypt request
+                        config.get(
+                            new VncKeyword(":encrypt"),
+	                        VncBoolean.False)));
+        }
+            catch(Exception ex) {
+                IO.safeClose(ch);
+                opened.set(false);
+                channel.set(null);
+                throw new VncException("Failed to get client config from server!", ex);
+            }
+
+            if (encrypt.get()) {
                 try {
                     diffieHellmanKeyExchange();
                 }
@@ -932,20 +937,27 @@ private IMessage sendAtomically(
             final Encryptor encryptor
     ) {
         try {
+            final boolean auditCount = msg.getType() != MessageType.DIFFIE_HELLMAN_KEY_REQUEST
+                                       && msg.getType() != MessageType.CLIENT_CONFIG;
+
             // sending the request message and receiving the response
             // must be atomic otherwise request and response can be mixed
             // in multi-threaded environments
             if (sendSemaphore.tryAcquire(120L, TimeUnit.SECONDS)) {
                 try {
                     Protocol.sendMessage(ch, (Message)msg, compressor, encryptor);
-                    messageSentCount.incrementAndGet();
+                    if (auditCount) {
+                        messageSentCount.incrementAndGet();
+                    }
 
                     if (msg.isOneway()) {
                         return null;
                     }
                     else {
                         final Message response = Protocol.receiveMessage(ch, compressor, encryptor);
-                        messageReceiveCount.incrementAndGet();
+                        if (auditCount) {
+                            messageReceiveCount.incrementAndGet();
+                        }
                         return response;
                     }
                 }
@@ -1141,6 +1153,25 @@ private static Message createQueuePollRequestMessage(
                 new byte[0]);
     }
 
+    private static Message createConfigRequestMessage() {
+        return new Message(
+                null,
+                null,
+                MessageType.CLIENT_CONFIG,
+                ResponseStatus.NULL,
+                false,
+                false,
+                null,
+                null,
+                System.currentTimeMillis(),
+                Message.EXPIRES_NEVER,
+                Message.NO_TIMEOUT,
+                Topics.of("client-config"),
+                "text/plain",
+                "UTF-8",
+                new byte[0]);
+    }
+
 
     private static class FutureNull implements Future<IMessage> {
 
@@ -1202,8 +1233,8 @@ private static byte[] toBytes(final String s, final String charset) {
     private final AtomicReference<Compressor> compressor = new AtomicReference<>(Compressor.off());
 
     // encryption
-    private final boolean encrypt;
     private final DiffieHellmanKeys dhKeys;
+    private final AtomicBoolean encrypt = new AtomicBoolean(false);
     private final AtomicReference<Encryptor> encryptor = new AtomicReference<>(Encryptor.off());
 
     private final ManagedCachedThreadPoolExecutor mngdExecutor =

src/main/java/com/github/jlangch/venice/util/ipc/TcpServer.java

@@ -95,6 +95,29 @@ public TcpServer setMaximumParallelConnections(final int count) {
         return this;
     }
 
+    /**
+     * Set the encryption mode
+     *
+     * @param encrypt if <code>true</code> encrypt the payload data at transport
+     *                level communication between this client and the server.
+     */
+    public void setEncryption(final boolean encrypt) {
+        if (started.get()) {
+            throw new VncException(
+                   "The encryption mode cannot be set anymore "
+                   + "once the server has been started!");
+        }
+        this.encrypt.set(encrypt);
+    }
+
+    /**
+     * @return <code>true</code> if this server has transport level encryption
+     *         enabled else <code>false</code>
+     */
+    public boolean isEncrypted() {
+        return encrypt.get();
+    }
+
     /**
      * Set the compression cutoff size for payload messages.
      *
@@ -504,6 +527,7 @@ private static byte[] toBytes(final String s, final String charset) {
     private final AtomicReference<ServerSocketChannel> server = new AtomicReference<>();
     private final AtomicLong maxMessageSize = new AtomicLong(MESSAGE_LIMIT_MAX);
     private final AtomicLong maxQueues = new AtomicLong(QUEUES_MAX);
+    private final AtomicBoolean encrypt = new AtomicBoolean(false);
     private final WalQueueManager wal = new WalQueueManager();
     private final int publishQueueCapacity = 50;
     private final ServerStatistics statistics = new ServerStatistics();