Add ability to limit outgoing message rate

.github/workflows/dev-docker-publish.yml

@@ -10,39 +10,10 @@ on:
       - README.md
 
 jobs:
-  build:
-    runs-on: ubuntu-20.04
-    steps:
-    - uses: actions/checkout@v2
-# Prepare custom build version
-    - name: Get branch name
-      id: branch
-      run: echo ::set-output name=branch_name::${GITHUB_REF#refs/*/}
-    - name: Get release_version
-      id: ver
-      uses: christian-draeger/[email protected]
-      with:
-        path: gradle.properties
-        property: release_version
-    - name: Build custom release version
-      id: release_ver
-      run: echo ::set-output name=value::"${{ steps.ver.outputs.value }}-${{ steps.branch.outputs.branch_name }}-${{ github.run_id }}"
-    - name: Show custom release version
-      run: echo ${{ steps.release_ver.outputs.value }}
-# Build and publish image
-    - name: Set up Docker Buildx
-      uses: docker/setup-buildx-action@v1
-    - uses: docker/login-action@v1
-      with:
-        registry: ghcr.io
-        username: ${{ github.repository_owner }}
-        password: ${{ secrets.CR_PAT }}
-    - run: echo "::set-output name=REPOSITORY_NAME::$(echo '${{ github.repository }}' | awk -F '/' '{print $2}')"
-      id: meta
-    - name: Build and push
-      id: docker_build
-      uses: docker/build-push-action@v2
-      with:
-        push: true
-        tags: ghcr.io/${{ github.repository }}:${{ steps.release_ver.outputs.value }}
-        labels: com.exactpro.th2.${{ steps.meta.outputs.REPOSITORY_NAME }}=${{ steps.ver.outputs.value }}
+  build-job:
+    uses: th2-net/.github/.github/workflows/compound-java-dev.yml@main
+    with:
+      build-target: 'Docker'
+      docker-username: ${{ github.actor }}
+    secrets:
+      docker-password: ${{ secrets.GITHUB_TOKEN }}

.github/workflows/docker-publish.yml

@@ -3,33 +3,17 @@ name: Build and publish Docker distributions to Github Container Registry ghcr.i
 on:
   push:
     branches:
-    - master
-    - version-*
+      - master
+      - version-*
+    paths:
+      - gradle.properties
+#    - package_info.json
 
 jobs:
-  build:
-    runs-on: ubuntu-20.04
-    steps:
-    - uses: actions/checkout@v2
-    - name: Set up Docker Buildx
-      uses: docker/setup-buildx-action@v1
-    - uses: docker/login-action@v1
-      with:
-        registry: ghcr.io
-        username: ${{ github.repository_owner }}
-        password: ${{ secrets.CR_PAT }}
-    - run: echo "::set-output name=REPOSITORY_NAME::$(echo '${{ github.repository }}' | awk -F '/' '{print $2}')"
-      id: meta
-    - name: Read version from gradle.properties
-      id: read_property
-      uses: christian-draeger/[email protected]
-      with:
-        path: ./gradle.properties
-        property: release_version
-    - name: Build and push
-      id: docker_build
-      uses: docker/build-push-action@v2
-      with:
-        push: true
-        tags: ghcr.io/${{ github.repository }}:${{ steps.read_property.outputs.value }}
-        labels: com.exactpro.th2.${{ steps.meta.outputs.REPOSITORY_NAME }}=${{ steps.read_property.outputs.value }}
+  build-job:
+    uses: th2-net/.github/.github/workflows/compound-java.yml@main
+    with:
+      build-target: 'Docker'
+      docker-username: ${{ github.actor }}
+    secrets:
+      docker-password: ${{ secrets.GITHUB_TOKEN }}

.gitlab-ci.yml

@@ -1,5 +1,5 @@
 variables:
-  APP_NAME: "th2-conn"
+  APP_NAME: "th2-conn-sailfish"
 
 include:
   - project: "vivarium/th2/pipelines"

README.md

@@ -1,4 +1,4 @@
-# Connect (3.11.0)
+# Connect (3.12.0)
 
 The "Connect" component is responsible for the communication with a target system.
 This component implements the logic of the interaction protocol, receiving and sending messages from and to the system, respectively.
@@ -28,6 +28,7 @@ Parameters:
 + name - the service name that will be displayed in the events inside the report;
 + settings - the parameters that will be transformed to the actual service's settings specified in the **services.xml** file.
 + maxMessageBatchSize - the limitation for message batch size which connect sends to the first and to the second publish pins with. The default value is set to 100.
++ maxMessageRate - max outgoing message rate in messages per second
 + maxMessageFlushTime - defines maximum time between outgoing message batches in milliseconds. The default value is set to 1000.
 + enableMessageSendingEvent - if this option is set to `true`, connect sends a separate event for every message sent which incomes from the pin with the send attribute. The default value is set to true
 
@@ -103,35 +104,42 @@ spec:
     type: "th2_service:Your_Service_Type"
     name: "your_service"
     maxMessageBatchSize: 100
+    maxMessageRate: 100000
     maxMessageFlushTime: 1000
     enableMessageSendingEvent: true
     settings:
       param1: "value1"
   pins:
     - name: in_raw
       connection-type: mq
-      attributes: ["first", "raw", "publish", "store"]
+      attributes: [ "first", "raw", "publish", "store" ]
     - name: out_raw
       connection-type: mq
-      attributes: ["second", "raw", "publish", "store"]
+      attributes: [ "second", "raw", "publish", "store" ]
     - name: to_send
       connection-type: mq
-      attributes: ["send", "raw", "subscribe"]
+      attributes: [ "send", "raw", "subscribe" ]
 ```
 
 ## Release notes
 
+### 3.12.0
++ Outgoing message rate now can be limited via `maxMessageRate` setting
+
 ### 3.11.0
 
++ Updated `sailfish-core` version from `3.2.1741` to `3.3.54`
++ Updated `common` from `3.33.0` to `3.44.0`
++ Updated `kotlin` form `1.5.30` to `1.6.21`
++ Renamed project to `conn-sailfish`
 + Added `maxMessageFlushTime` option
-+ Dependencies with vulnerabilities were updated:
-  + Sailfish updated from `3.2.1741` to `3.3.13`
 
 ### 3.10.2
 
 + Events are made more convenient. Added event names and error logs. Error message moved from the name to the body of the event.
 + Use temporal directory for last layer in sailfish's workspace
 
+
 ### 3.10.1
 
 + Update `sailfish-core` version from `3.2.1674` to `3.2.1741`

build.gradle

@@ -2,26 +2,45 @@ plugins {
     id 'com.palantir.docker' version '0.25.0'
     id 'org.jetbrains.kotlin.jvm' version '1.6.21'
     id 'application'
+    id "org.owasp.dependencycheck" version "8.2.1"
+}
+
+dependencyCheck {
+    formats=['SARIF', 'JSON', 'HTML']
+    failBuildOnCVSS=5
+    analyzers {
+        assemblyEnabled = false
+        nugetconfEnabled = false
+        nodeEnabled = false
+    }
 }
 
 ext {
-    sharedDir                       = file("${project.rootDir}/shared")
-    sailfishVersion                 = '3.3.13'
+    sailfishVersion = '3.3.122'
+}
+
+ext.excludeSailfish = { rcd ->
+    rcd.excludeModule("com.exactpro.sf", "sailfish-core")
+    rcd.excludeModule("com.exactpro.sf", "sailfish-common")
+    rcd.excludeModule("com.exactpro.sf", "sailfish-rest-api-client")
+    rcd.excludeModule("com.exactpro.sf", "service-http")
 }
 
 group = 'com.exactpro.th2'
 version = release_version
 
 repositories {
     maven {
-        name 'MavenLocal'
-        url sharedDir
+        name 'Sonatype_snapshots'
+        url 'https://s01.oss.sonatype.org/content/repositories/snapshots/'
+        content {
+            excludeSailfish(it)
+        }
     }
-
     // ignoreGradleMetadataRedirection is used for sonatype because
     // Sailfish dependencies have constrains that interfere with our BOM
-    // so we exclude Gradle metadata for this repositories
-
+    // so we exclude Gradle metadata for this repositories.
+    // We've checked these versions - they are compatible and safe to use
     maven {
         name 'Sonatype_snapshots'
         url 'https://s01.oss.sonatype.org/content/repositories/snapshots/'
@@ -34,18 +53,20 @@ repositories {
     maven {
         name 'Sonatype_releases'
         url 'https://s01.oss.sonatype.org/content/repositories/releases/'
-        metadataSources {
-            mavenPom()
-            artifact()
-            ignoreGradleMetadataRedirection()
+        content {
+            excludeSailfish(it)
         }
     }
-    mavenCentral {
+    maven {
+        name 'Sonatype_releases'
+        url 'https://s01.oss.sonatype.org/content/repositories/releases/'
         metadataSources {
             mavenPom()
             artifact()
+            ignoreGradleMetadataRedirection()
         }
     }
+    mavenCentral()
     mavenLocal()
 
     configurations.all {
@@ -80,22 +101,17 @@ compileTestKotlin {
 }
 
 dependencies {
-    api platform('com.exactpro.th2:bom:4.0.1')
+    api platform('com.exactpro.th2:bom:4.1.0')
 
-    implementation 'com.exactpro.th2:common:3.42.0-dev-3174619150-SNAPSHOT'
-    implementation 'com.exactpro.th2:sailfish-utils:3.13.0-dev-3204703314-SNAPSHOT'
-
-    implementation "org.slf4j:slf4j-api"
+    implementation 'com.exactpro.th2:common:3.44.0'
+    implementation 'com.exactpro.th2:sailfish-utils:3.15.0'
 
     implementation group: 'io.reactivex.rxjava3', name: 'rxjava', version: '3.0.4'
 
     implementation 'io.prometheus:simpleclient'
 
     implementation("com.exactpro.sf:sailfish-core:${sailfishVersion}")
 
-    testImplementation('org.apache.logging.log4j:log4j-slf4j-impl') {
-        because('logging in testing')
-    }
     testImplementation 'org.junit.jupiter:junit-jupiter:5.9.0'
 }
 

gradle.properties

@@ -1 +1,5 @@
+<<<<<<< HEAD
+release_version=3.12.0
+=======
 release_version=3.11.0
+>>>>>>> dev

settings.gradle

@@ -1 +1 @@
-rootProject.name = 'conn'
+rootProject.name = 'conn-sailfish'

src/main/java/com/exactpro/th2/conn/ConnectivityMessage.java

@@ -16,6 +16,7 @@
 package com.exactpro.th2.conn;
 
 import static com.exactpro.sf.common.messages.MetadataExtensions.getMessageProperties;
+import static com.exactpro.th2.common.message.MessageUtils.toTimestamp;
 import static com.google.protobuf.TextFormat.shortDebugString;
 import static java.util.Objects.requireNonNull;
 import static org.apache.commons.lang3.ObjectUtils.defaultIfNull;
@@ -46,8 +47,6 @@
 
 public class ConnectivityMessage {
     private static final Logger LOGGER = LoggerFactory.getLogger(ConnectivityMessage.class);
-    public static final long MILLISECONDS_IN_SECOND = 1_000L;
-    public static final long NANOSECONDS_IN_MILLISECOND = 1_000_000L;
 
     private final List<IMessage> sailfishMessages;
 
@@ -62,7 +61,7 @@ public ConnectivityMessage(List<IMessage> sailfishMessages, String sessionAlias,
         }
         messageID = createMessageID(createConnectionID(requireNonNull(sessionAlias, "Session alias can't be null")),
                 requireNonNull(direction, "Direction can't be null"), sequence);
-        timestamp = createTimestamp(sailfishMessages.get(0).getMetaData().getMsgTimestamp().getTime());
+        timestamp = toTimestamp(sailfishMessages.get(0).getMetaData().getPreciseMsgTimestamp());
     }
 
     public String getSessionAlias() {
@@ -158,12 +157,4 @@ private static RawMessageMetadata.Builder createRawMessageMetadataBuilder(Messag
                 .setId(messageID)
                 .setTimestamp(timestamp);
     }
-
-    // TODO: Required nanosecond accuracy
-    private static Timestamp createTimestamp(long milliseconds) {
-        return Timestamp.newBuilder()
-                .setSeconds(milliseconds / MILLISECONDS_IN_SECOND)
-                .setNanos((int) (milliseconds % MILLISECONDS_IN_SECOND * NANOSECONDS_IN_MILLISECOND))
-                .build();
-    }
 }

src/main/java/com/exactpro/th2/conn/MessageSender.java

@@ -15,15 +15,6 @@
  */
 package com.exactpro.th2.conn;
 
-import static java.util.Objects.requireNonNull;
-
-import java.io.IOException;
-import java.util.Base64;
-import java.util.Map;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import com.exactpro.sf.common.messages.IMetadata;
 import com.exactpro.sf.common.messages.MetadataExtensions;
 import com.exactpro.sf.common.messages.impl.Metadata;
@@ -40,8 +31,15 @@
 import com.exactpro.th2.conn.events.EventHolder;
 import com.exactpro.th2.conn.utility.EventStoreExtensions;
 import com.exactpro.th2.conn.utility.SailfishMetadataExtensions;
-
 import io.reactivex.rxjava3.annotations.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Base64;
+import java.util.Map;
+
+import static java.util.Objects.requireNonNull;
 
 public class MessageSender {
     private static final String SEND_ATTRIBUTE = "send";
@@ -50,16 +48,26 @@ public class MessageSender {
     private final MessageRouter<RawMessageBatch> router;
     private final EventDispatcher eventDispatcher;
     private final EventID untrackedMessagesRoot;
+    private final RateLimiter rateLimiter;
     private volatile SubscriberMonitor subscriberMonitor;
 
     public MessageSender(IServiceProxy serviceProxy,
                          MessageRouter<RawMessageBatch> router,
                          EventDispatcher eventDispatcher,
                          EventID untrackedMessagesRoot) {
+        this(serviceProxy, router, eventDispatcher, untrackedMessagesRoot, Integer.MAX_VALUE);
+    }
+
+    public MessageSender(IServiceProxy serviceProxy,
+                         MessageRouter<RawMessageBatch> router,
+                         EventDispatcher eventDispatcher,
+                         EventID untrackedMessagesRoot,
+                         int maxMessageRate) {
         this.serviceProxy = requireNonNull(serviceProxy, "Service proxy can't be null");
         this.router = requireNonNull(router, "Message router can't be null");
         this.eventDispatcher = requireNonNull(eventDispatcher, "'Event dispatcher' can't be null");
         this.untrackedMessagesRoot = requireNonNull(untrackedMessagesRoot, "'untrackedMessagesRoot' can't be null");
+        this.rateLimiter = new RateLimiter(maxMessageRate);
     }
 
     public void start() {
@@ -86,6 +94,7 @@ public void stop() throws IOException {
     private void handle(String consumerTag, RawMessageBatch messageBatch) {
         for (RawMessage protoMessage : messageBatch.getMessagesList()) {
             try {
+                rateLimiter.acquire();
                 sendMessage(protoMessage);
             } catch (InterruptedException e) {
                 logger.error("Send message operation interrupted. Consumer tag {}", consumerTag, e);