Channel uri socket params

aeron-client/src/main/c/uri/aeron_uri.c

@@ -437,6 +437,28 @@ int aeron_uri_get_ats(aeron_uri_params_t *uri_params, aeron_uri_ats_status_t *ur
     return 1;
 }
 
+int aeron_uri_get_socket_bufs(aeron_uri_params_t *uri_params, size_t *socket_sndbuf, size_t *socket_rcvbuf)
+{
+    int64_t socket_sndbuf_tmp = 0;
+    if (aeron_uri_get_int64(uri_params, AERON_URI_SOCKET_SNDBUF_KEY, &socket_sndbuf_tmp) < 0)
+    {
+        AERON_SET_ERR(EINVAL, "%s", "Failed to parse socket-sndbuf");
+        return -1;
+    }
+
+    int64_t socket_rcvbuf_tmp = 0;
+    if (aeron_uri_get_int64(uri_params, AERON_URI_SOCKET_RCVBUF_KEY, &socket_rcvbuf_tmp) < 0)
+    {
+        AERON_SET_ERR(EINVAL, "%s", "Failed to parse socket-rcvbuf");
+        return -1;
+    }
+
+    *socket_sndbuf = (size_t)socket_sndbuf_tmp;
+    *socket_rcvbuf = (size_t)socket_rcvbuf_tmp;
+
+    return 0;
+}
+
 int64_t aeron_uri_parse_tag(const char *tag_str)
 {
     errno = 0;

aeron-client/src/main/c/uri/aeron_uri.h

@@ -62,6 +62,8 @@ aeron_uri_params_t;
 #define AERON_URI_CC_KEY "cc"
 #define AERON_URI_SPIES_SIMULATE_CONNECTION_KEY "ssc"
 #define AERON_URI_ATS_KEY "ats"
+#define AERON_URI_SOCKET_SNDBUF_KEY "socket-sndbuf"
+#define AERON_URI_SOCKET_RCVBUF_KEY "socket-rcvbuf"
 
 #define AERON_URI_INVALID_TAG (-1)
 
@@ -133,6 +135,7 @@ int aeron_uri_get_int64(aeron_uri_params_t *uri_params, const char *key, int64_t
 int aeron_uri_get_bool(aeron_uri_params_t *uri_params, const char *key, bool *retval);
 int aeron_uri_get_ats(aeron_uri_params_t *uri_params, aeron_uri_ats_status_t *uri_ats_status);
 int aeron_uri_sprint(aeron_uri_t *uri, char *buffer, size_t buffer_len);
+int aeron_uri_get_socket_bufs(aeron_uri_params_t *uri_params, size_t *socket_sndbuf, size_t *socket_rcvbuf);
 
 int64_t aeron_uri_parse_tag(const char *tag_str);
 

aeron-client/src/main/cpp_wrapper/ChannelUri.h

@@ -60,6 +60,8 @@ constexpr const char CONGESTION_CONTROL_PARAM_NAME[] = "cc";
 constexpr const char FLOW_CONTROL_PARAM_NAME[] = "fc";
 constexpr const char GROUP_TAG_PARAM_NAME[] = "gtag";
 constexpr const char SPIES_SIMULATE_CONNECTION_PARAM_NAME[] = "ssc";
+constexpr const char SOCKET_SNDBUF_PARAM_NAME[] = "socket-sndbuf";
+constexpr const char SOCKET_RCVBUF_PARAM_NAME[] = "socket-rcvbuf";
 
 using namespace aeron::util;
 

aeron-client/src/main/cpp_wrapper/ChannelUriStringBuilder.h

@@ -62,6 +62,7 @@ class ChannelUriStringBuilder
         m_group.reset(nullptr);
         m_rejoin.reset(nullptr);
         m_ssc.reset(nullptr);
+
         m_isSessionIdTagged = false;
         return *this;
     }
@@ -295,6 +296,30 @@ class ChannelUriStringBuilder
         return *this;
     }
 
+    inline this_t &socketSndbufLength(std::uint32_t socketSndbufLength)
+    {
+        m_socketSndbufLength.reset(new Value(socketSndbufLength));
+        return *this;
+    }
+
+    inline this_t &socketSndbufLength(std::nullptr_t socketSndbufLength)
+    {
+        m_socketSndbufLength.reset(nullptr);
+        return *this;
+    }
+
+    inline this_t &socketRcvbufLength(std::uint32_t socketRcvbufLength)
+    {
+        m_socketRcvbufLength.reset(new Value(socketRcvbufLength));
+        return *this;
+    }
+
+    inline this_t &socketRcvbufLength(std::nullptr_t)
+    {
+        m_socketRcvbufLength.reset(nullptr);
+        return *this;
+    }
+
     inline this_t &initialPosition(std::int64_t position, std::int32_t initialTermId, std::int32_t termLength)
     {
         if (position < 0 || 0 != (position & (aeron::concurrent::logbuffer::FrameDescriptor::FRAME_ALIGNMENT - 1)))
@@ -445,6 +470,16 @@ class ChannelUriStringBuilder
             sb << SPIES_SIMULATE_CONNECTION_PARAM_NAME << '=' << (m_ssc->value == 1 ? "true" : "false") << '|';
         }
 
+        if (m_socketSndbufLength)
+        {
+            sb << SOCKET_SNDBUF_PARAM_NAME << '=' << m_socketSndbufLength->value << '|';
+        }
+
+        if (m_socketRcvbufLength)
+        {
+            sb << SOCKET_RCVBUF_PARAM_NAME << '=' << m_socketRcvbufLength->value << '|';
+        }
+
         std::string result = sb.str();
         const char lastChar = result.back();
 
@@ -493,6 +528,8 @@ class ChannelUriStringBuilder
     std::unique_ptr<Value> m_group;
     std::unique_ptr<Value> m_rejoin;
     std::unique_ptr<Value> m_ssc;
+    std::unique_ptr<Value> m_socketSndbufLength;
+    std::unique_ptr<Value> m_socketRcvbufLength;
     bool m_isSessionIdTagged = false;
 
     inline static std::string prefixTag(bool isTagged, Value &value)

aeron-client/src/main/java/io/aeron/ChannelUriStringBuilder.java

@@ -68,6 +68,8 @@ public final class ChannelUriStringBuilder
     private Boolean rejoin;
     private Boolean ssc;
     private boolean isSessionIdTagged;
+    private Integer socketSndbufLength;
+    private Integer socketRcvbufLength;
 
     /**
      * Clear out all the values thus setting back to the initial state.
@@ -102,6 +104,8 @@ public ChannelUriStringBuilder clear()
         group = null;
         rejoin = null;
         isSessionIdTagged = false;
+        socketRcvbufLength = null;
+        socketSndbufLength = null;
 
         return this;
     }
@@ -1441,6 +1445,83 @@ public ChannelUriStringBuilder initialPosition(final long position, final int in
         return this;
     }
 
+    /**
+     * Set the underlying OS send buffer length.
+     *
+     * @param socketSndbufLength paramter to be passed as SO_SNDBUF value
+     * @return this for a fluent API.
+     * @see CommonContext#SOCKET_SNDBUF_PARAM_NAME
+     */
+    public ChannelUriStringBuilder socketSndbufLength(final Integer socketSndbufLength)
+    {
+        this.socketSndbufLength = socketSndbufLength;
+        return this;
+    }
+
+    /**
+     * Set the underlying OS send buffer length from an existing {@link ChannelUri} which may be (null).
+     *
+     * @param channelUri to read the value from.
+     * @return this for a fluent API.
+     * @see CommonContext#SOCKET_SNDBUF_PARAM_NAME
+     */
+    public ChannelUriStringBuilder socketSndbufLength(final ChannelUri channelUri)
+    {
+        final String socketSndbufLengthString = channelUri.get(SOCKET_SNDBUF_PARAM_NAME);
+        this.socketSndbufLength = null == socketSndbufLengthString ? null : Integer.valueOf(socketSndbufLengthString);
+        return this;
+    }
+
+    /**
+     * Get the underling OS send buffer length setting
+     *
+     * @return underlying OS send buffer length setting or null if not specified.
+     * @see CommonContext#SOCKET_SNDBUF_PARAM_NAME
+     */
+    public Integer socketSndbufLength()
+    {
+        return socketSndbufLength;
+    }
+
+
+    /**
+     * Set the underlying OS receive buffer length.
+     *
+     * @param socketRcvbufLength paramter to be passed as SO_SNDBUF value
+     * @return this for a fluent API.
+     * @see CommonContext#SOCKET_RCVBUF_PARAM_NAME
+     */
+    public ChannelUriStringBuilder socketRcvbufLength(final Integer socketRcvbufLength)
+    {
+        this.socketRcvbufLength = socketRcvbufLength;
+        return this;
+    }
+
+    /**
+     * Set the underlying OS receive buffer length from an existing {@link ChannelUri} which may be (null).
+     *
+     * @param channelUri to read the value from.
+     * @return this for a fluent API.
+     * @see CommonContext#SOCKET_RCVBUF_PARAM_NAME
+     */
+    public ChannelUriStringBuilder socketRcvbufLength(final ChannelUri channelUri)
+    {
+        final String socketRcvbufLengthString = channelUri.get(SOCKET_RCVBUF_PARAM_NAME);
+        this.socketRcvbufLength = null == socketRcvbufLengthString ? null : Integer.valueOf(socketRcvbufLengthString);
+        return this;
+    }
+
+    /**
+     * Get the underling OS receive buffer length setting
+     *
+     * @return underlying OS receive buffer length setting or null if not specified.
+     * @see CommonContext#SOCKET_RCVBUF_PARAM_NAME
+     */
+    public Integer socketRcvbufLength()
+    {
+        return socketRcvbufLength;
+    }
+
     /**
      * Build a channel URI String for the given parameters.
      *
@@ -1578,6 +1659,16 @@ public String build()
             sb.append(SPIES_SIMULATE_CONNECTION_PARAM_NAME).append('=').append(ssc).append('|');
         }
 
+        if (null != socketSndbufLength)
+        {
+            sb.append(SOCKET_SNDBUF_PARAM_NAME).append('=').append(socketSndbufLength).append('|');
+        }
+
+        if (null != socketRcvbufLength)
+        {
+            sb.append(SOCKET_RCVBUF_PARAM_NAME).append('=').append(socketRcvbufLength).append('|');
+        }
+
         final char lastChar = sb.charAt(sb.length() - 1);
         if (lastChar == '|' || lastChar == '?')
         {

aeron-client/src/main/java/io/aeron/CommonContext.java

@@ -304,6 +304,16 @@ public static InferableBoolean parse(final String value)
      */
     public static final String SPIES_SIMULATE_CONNECTION_PARAM_NAME = "ssc";
 
+    /**
+     * Parameter name for the underlying OS socket send buffer size
+     */
+    public static final String SOCKET_SNDBUF_PARAM_NAME = "socket-sndbuf";
+
+    /**
+     * Parameter name for the underlying OS socket receive buffer size
+     */
+    public static final String SOCKET_RCVBUF_PARAM_NAME = "socket-rcvbuf";
+
     /**
      * Using an integer because there is no support for boolean. 1 is concluded, 0 is not concluded.
      */

aeron-client/src/test/cpp_wrapper/CMakeLists.txt

@@ -61,3 +61,4 @@ aeron_client_wrapper_test(livenessTimeoutTest LivenessTimeoutTest.cpp)
 aeron_client_native_test(livenessTimeoutTest LivenessTimeoutTest.cpp)
 
 aeron_client_wrapper_test(exceptionsTest ExceptionsTest.cpp)
+aeron_client_wrapper_test(channelUriStringBuilderTest ChannelUriStringBuilderTest.cpp)

aeron-client/src/test/cpp_wrapper/ChannelUriStringBuilderTest.cpp

@@ -0,0 +1,115 @@
+/*
+ * Copyright 2014-2021 Real Logic Limited.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "gtest/gtest.h"
+
+#include "ChannelUriStringBuilder.h"
+
+using namespace aeron;
+
+TEST(ChannelUriStringBuilderTest, shouldGenerateBasicIpcChannel)
+{
+    ChannelUriStringBuilder builder;
+
+    builder.media(IPC_MEDIA);
+
+    ASSERT_EQ(builder.build(), "aeron:ipc");
+}
+
+TEST(ChannelUriStringBuilderTest, shouldGenerateBasicUdpChannel)
+{
+    ChannelUriStringBuilder builder;
+
+    builder
+        .media(UDP_MEDIA)
+        .endpoint("localhost:9999");
+
+    ASSERT_EQ(builder.build(), "aeron:udp?endpoint=localhost:9999");
+}
+
+TEST(ChannelUriStringBuilderTest, shouldGenerateBasicUdpChannelSpy)
+{
+    ChannelUriStringBuilder builder;
+
+    builder
+        .prefix(SPY_QUALIFIER)
+        .media(UDP_MEDIA)
+        .endpoint("localhost:9999");
+
+    ASSERT_EQ(builder.build(), "aeron-spy:aeron:udp?endpoint=localhost:9999");
+}
+
+TEST(ChannelUriStringBuilderTest, shouldGenerateComplexUdpChannel)
+{
+    ChannelUriStringBuilder builder;
+
+    builder
+        .media(UDP_MEDIA)
+        .endpoint("localhost:9999")
+        .ttl(9)
+        .termLength(1024 * 128);
+
+    ASSERT_EQ(builder.build(), "aeron:udp?endpoint=localhost:9999|term-length=131072|ttl=9");
+}
+
+TEST(ChannelUriStringBuilderTest, shouldGenerateReplayUdpChannel)
+{
+    ChannelUriStringBuilder builder;
+
+    builder
+        .media(UDP_MEDIA)
+        .endpoint("localhost:9999")
+        .termLength(1024 * 128)
+        .initialTermId(777)
+        .termId(999)
+        .termOffset(64);
+
+    ASSERT_EQ(
+        builder.build(),
+        "aeron:udp?endpoint=localhost:9999|term-length=131072|init-term-id=777|term-id=999|term-offset=64");
+}
+
+TEST(ChannelUriStringBuilderTest, shouldGenerateInitialPosition)
+{
+    ChannelUriStringBuilder builder;
+
+    std::uint32_t termLength = 1024 * 128;
+    std::int64_t position = (termLength * 3) + 64;
+
+    builder
+        .media(UDP_MEDIA)
+        .endpoint("localhost:9999")
+        .initialPosition(position, 777, termLength);
+
+    ASSERT_EQ(
+        builder.build(),
+        "aeron:udp?endpoint=localhost:9999|term-length=131072|init-term-id=777|term-id=780|term-offset=64");
+}
+
+TEST(ChannelUriStringBuilderTest, shouldGenerateSocketSndRcvbufLengths)
+{
+    ChannelUriStringBuilder builder;
+
+    builder
+        .media(UDP_MEDIA)
+        .endpoint("localhost:9999")
+        .socketSndbufLength(8192)
+        .socketRcvbufLength(4096);
+
+    ASSERT_EQ(
+        builder.build(),
+        "aeron:udp?endpoint=localhost:9999|socket-sndbuf=8192|socket-rcvbuf=4096");
+}

aeron-client/src/test/java/io/aeron/ChannelUriStringBuilderTest.java

@@ -100,4 +100,18 @@ public void shouldGenerateReplayUdpChannel()
             "aeron:udp?endpoint=address:9999|term-length=131072|init-term-id=777|term-id=999|term-offset=64",
             builder.build());
     }
-}
+
+    @Test
+    public void shouldGenerateChannelWithSocketParameters()
+    {
+        final ChannelUriStringBuilder builder = new ChannelUriStringBuilder()
+            .media("udp")
+            .endpoint("address:9999")
+            .socketSndbufLength(8192)
+            .socketRcvbufLength(4096);
+
+        assertEquals(
+            "aeron:udp?endpoint=address:9999|socket-sndbuf=8192|socket-rcvbuf=4096",
+            builder.build());
+    }
+}