Propagate client properties

Author: acogoluegnes

pom.xml

@@ -179,6 +179,14 @@
 
     <build>
 
+        <!-- We store the client version in a Java properties file. -->
+        <resources>
+            <resource>
+                <directory>${basedir}/src/main/resources</directory>
+                <filtering>true</filtering>
+            </resource>
+        </resources>
+
         <plugins>
 
             <plugin>

src/main/java/com/rabbitmq/stream/Client.java

@@ -97,6 +97,8 @@ public class Client implements AutoCloseable {
 
     private final ChunkChecksum chunkChecksum;
 
+    private final Map<String, String> clientProperties;
+
     private final String NETTY_HANDLER_FLUSH_CONSOLIDATION = FlushConsolidationHandler.class.getSimpleName();
     private final String NETTY_HANDLER_FRAME_DECODER = LengthFieldBasedFrameDecoder.class.getSimpleName();
     private final String NETTY_HANDLER_STREAM = StreamHandler.class.getSimpleName();
@@ -154,6 +156,7 @@ public void initChannel(SocketChannel ch) {
 
         this.channel = f.channel();
         this.tuneState = new TuneState(parameters.requestedMaxFrameSize, (int) parameters.requestedHeartbeat.getSeconds());
+        this.clientProperties = clientProperties(parameters.clientProperties);
         authenticate();
         this.tuneState.await(Duration.ofSeconds(10));
         this.maxFrameSize = this.tuneState.getMaxFrameSize();
@@ -163,6 +166,13 @@ public void initChannel(SocketChannel ch) {
         open(parameters.virtualHost);
     }
 
+    private static Map<String, String> clientProperties(Map<String, String> fromParameters) {
+        fromParameters = fromParameters == null ? Collections.emptyMap() : fromParameters;
+        Map<String, String> clientProperties = new HashMap<>(fromParameters);
+        clientProperties.putAll(ClientProperties.DEFAULT_CLIENT_PROPERTIES);
+        return Collections.unmodifiableMap(clientProperties);
+    }
+
     static void handleMetadataUpdate(ByteBuf bb, int frameSize, Subscriptions subscriptions, SubscriptionListener subscriptionListener) {
         int read = 2 + 2; // already read the command id and version
         short code = bb.readShort();
@@ -573,7 +583,7 @@ private void authenticate() {
         boolean authDone = false;
         while (!authDone) {
             byte[] saslResponse = saslMechanism.handleChallenge(challenge, this.credentialsProvider);
-            SaslAuthenticateResponse saslAuthenticateResponse = sendSaslAuthenticate(saslMechanism, saslResponse);
+            SaslAuthenticateResponse saslAuthenticateResponse = sendSaslAuthenticate(this.clientProperties, saslMechanism, saslResponse);
             if (saslAuthenticateResponse.isOk()) {
                 authDone = true;
             } else if (saslAuthenticateResponse.isChallenge()) {
@@ -586,8 +596,14 @@ private void authenticate() {
         }
     }
 
-    private SaslAuthenticateResponse sendSaslAuthenticate(SaslMechanism saslMechanism, byte[] challengeResponse) {
-        int length = 2 + 2 + 4 + 2 + saslMechanism.getName().length() +
+    private SaslAuthenticateResponse sendSaslAuthenticate(Map<String, String> clientProperties, SaslMechanism saslMechanism, byte[] challengeResponse) {
+        int clientPropertiesSize = 4; // size of the map, always there
+        if (!clientProperties.isEmpty()) {
+            for (Map.Entry<String, String> entry : clientProperties.entrySet()) {
+                clientPropertiesSize += 2 + entry.getKey().length() + 2 + entry.getValue().length();
+            }
+        }
+        int length = 2 + 2 + 4 + +clientPropertiesSize + 2 + saslMechanism.getName().length() +
                 4 + (challengeResponse == null ? 0 : challengeResponse.length);
         int correlationId = correlationSequence.incrementAndGet();
         try {
@@ -596,6 +612,11 @@ private SaslAuthenticateResponse sendSaslAuthenticate(SaslMechanism saslMechanis
             bb.writeShort(COMMAND_SASL_AUTHENTICATE);
             bb.writeShort(VERSION_0);
             bb.writeInt(correlationId);
+            bb.writeInt(clientProperties.size());
+            for (Map.Entry<String, String> entry : clientProperties.entrySet()) {
+                bb.writeShort(entry.getKey().length()).writeBytes(entry.getKey().getBytes(StandardCharsets.UTF_8))
+                        .writeShort(entry.getValue().length()).writeBytes(entry.getValue().getBytes(StandardCharsets.UTF_8));
+            }
             bb.writeShort(saslMechanism.getName().length());
             bb.writeBytes(saslMechanism.getName().getBytes(StandardCharsets.UTF_8));
             if (challengeResponse == null) {
@@ -1268,6 +1289,8 @@ public static class ClientParameters {
 
         private ChunkChecksum chunkChecksum = JdkChunkChecksum.CRC32_SINGLETON;
 
+        private final Map<String, String> clientProperties = new HashMap<>();
+
         public ClientParameters host(String host) {
             this.host = host;
             return this;
@@ -1371,6 +1394,16 @@ public ClientParameters chunkChecksum(ChunkChecksum chunkChecksum) {
             this.chunkChecksum = chunkChecksum;
             return this;
         }
+
+        public ClientParameters clientProperties(Map<String, String> clientProperties) {
+            this.clientProperties.putAll(clientProperties);
+            return this;
+        }
+
+        public ClientParameters clientProperty(String key, String value) {
+            this.clientProperties.put(key, value);
+            return this;
+        }
     }
 
     private static final class BinaryOnlyMessage implements Message {

src/main/java/com/rabbitmq/stream/ClientProperties.java

@@ -0,0 +1,91 @@
+// Copyright (c) 2020 VMware, Inc. or its affiliates.  All rights reserved.
+//
+// This software, the RabbitMQ Java client library, is dual-licensed under the
+// Mozilla Public License 1.1 ("MPL"), and the Apache License version 2 ("ASL").
+// For the MPL, please see LICENSE-MPL-RabbitMQ. For the ASL,
+// please see LICENSE-APACHE2.
+//
+// This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND,
+// either express or implied. See the LICENSE file for specific language governing
+// rights and limitations of this software.
+//
+// If you have any questions regarding licensing, please contact us at
+// [email protected].
+
+package com.rabbitmq.stream;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.InputStream;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+public abstract class ClientProperties {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(ClientProperties.class);
+
+    // We store the version property in an unusual way because relocating the package can rewrite the key in the property
+    // file, which results in spurious warnings being emitted at start-up.
+    // see https://github.com/rabbitmq/rabbitmq-java-client/issues/436
+    private static final char[] VERSION_PROPERTY = new char[]{'c', 'o', 'm', '.', 'r', 'a', 'b', 'b', 'i', 't', 'm', 'q', '.',
+            's', 't', 'r', 'e', 'a', 'm', '.', 'c', 'l', 'i', 'e', 'n', 't', '.', 'v', 'e', 'r', 's', 'i', 'o', 'n'};
+
+    public static final String VERSION = getVersion();
+
+    public static final Map<String, String> DEFAULT_CLIENT_PROPERTIES = Collections.unmodifiableMap(new HashMap<String, String>() {{
+        put("product", "RabbitMQ Stream");
+        put("version", ClientProperties.VERSION);
+        put("platform", "Java");
+        put("copyright", "Copyright (c) 2020 VMware, Inc. or its affiliates.");
+        put("information", "Licensed under the MPL 2.0. See https://www.rabbitmq.com/");
+    }});
+
+    private static String getVersion() {
+        String version;
+        try {
+            version = getVersionFromPropertyFile();
+        } catch (Exception e1) {
+            LOGGER.warn("Couldn't get version from property file", e1);
+            try {
+                version = getVersionFromPackage();
+            } catch (Exception e2) {
+                LOGGER.warn("Couldn't get version with Package#getImplementationVersion", e1);
+                version = getDefaultVersion();
+            }
+        }
+        return version;
+    }
+
+    private static String getVersionFromPropertyFile() throws Exception {
+        InputStream inputStream = ClientProperties.class.getClassLoader().getResourceAsStream("rabbitmq-stream-client.properties");
+        java.util.Properties version = new Properties();
+        try {
+            version.load(inputStream);
+        } finally {
+            if (inputStream != null) {
+                inputStream.close();
+            }
+        }
+        String propertyName = new String(VERSION_PROPERTY);
+        String versionProperty = version.getProperty(propertyName);
+        if (versionProperty == null) {
+            throw new IllegalStateException("Couldn't find version property in property file");
+        }
+        return versionProperty;
+    }
+
+    private static String getVersionFromPackage() {
+        if (ClientProperties.class.getPackage().getImplementationVersion() == null) {
+            throw new IllegalStateException("Couldn't get version with Package#getImplementationVersion");
+        }
+        return ClientProperties.class.getPackage().getImplementationVersion();
+    }
+
+    private static String getDefaultVersion() {
+        return "0.0.0";
+    }
+
+}

src/main/resources/rabbitmq-stream-client.properties

@@ -0,0 +1 @@
+com.rabbitmq.stream.client.version = ${project.version}