[TS-981] Update project version and return backward compatibility

README.md

@@ -12,8 +12,6 @@ The project contains the implementation of an AMQP connection **th2-conn-amqp**.
 > jks&trustStorePassword=<trustStorePassword>&keyStorePath=<keyStorePath>.
 > jks&keyStorePassword=<keyStorePassword>&saslMechanisms=EXTERNAL
 
-* username = \<username\>
-* password = \<password\>
 * sendQueue = \<sendQueue\>
 * receiveQueue = \<receiveQueue\>
 
@@ -50,6 +48,17 @@ spec:
       # factorylookup:
       # sendQueue:
       # receiveQueue:
+
+# Deprecated and will be deleted in the next major version.
+# Instead of sessions there is a way to use parameters with sessionAlias for single connection
+# But you can use only one of these ways
+  sessionAlias: session-alias
+  parameters:
+        # initialContextFactory:
+        # factorylookup:
+        # sendQueue:
+        # receiveQueue:
+
   pins:
     - name: in_raw
       connection-type: mq
@@ -60,4 +69,11 @@ spec:
     - name: to_send
       connection-type: mq
       attributes: ["send", "parsed", "subscribe"]
-```
+```
+
+# Release notes
+## 1.2.0
++ Add possibility of multiple connections
+    + Add list of sessions with connection parameters
+    + Backward compatibility remains
+

gradle.properties

@@ -1,6 +1,6 @@
 kotlin.code.style=official
 
-release_version=1.1.0
+release_version=1.2.0
 kotlin_version=1.5.31
 description = 'AMQP Client'
 

src/main/kotlin/com/exactpro/th2/conn/amqp/BoxMain.kt

@@ -27,6 +27,7 @@ import com.exactpro.th2.common.metrics.readiness
 import com.exactpro.th2.common.schema.factory.CommonFactory
 import com.exactpro.th2.common.schema.message.MessageRouter
 import com.exactpro.th2.conn.amqp.configuration.Configuration
+import com.exactpro.th2.conn.amqp.configuration.ConnParameters
 import com.exactpro.th2.conn.amqp.connservice.ConnService
 import com.exactpro.th2.conn.amqp.connservice.ConnServiceImpl
 import mu.KotlinLogging
@@ -69,10 +70,12 @@ fun main(args: Array<String>) {
                 .build()
         )
 
+        configureConnectionParameters(configuration, eventRouter, rootEvent)
+
         val rawRouter: MessageRouter<RawMessageBatch> = factory.messageRouterRawBatch
         val aliasToService = HashMap<String, ConnService>()
         val executor: ScheduledExecutorService = Executors.newSingleThreadScheduledExecutor()
-        configuration.sessions.forEach { connParameters ->
+        configuration.sessions?.forEach { connParameters ->
             run {
                 val publisher = MessagePublisher(
                     connParameters.sessionAlias,
@@ -92,15 +95,16 @@ fun main(args: Array<String>) {
                 )
                 resources += service
                 aliasToService[connParameters.sessionAlias] = service
+                service.start()
             }
-        }
+        } ?: error("Connection parameters can't be blank. Use {sessions} or {parameters + sessioAlias}")
         rawRouter.subscribeAll { _, rawBatch ->
             rawBatch.messagesList.forEach { msg ->
                 msg.runCatching {
                     val alias = msg.metadata.id.connectionId.sessionAlias
                     aliasToService.getOrElse(
                         alias,
-                        {throw IllegalArgumentException("Can't find service by alias {$alias}")}
+                        { error("Can't find service by alias {$alias}") }
                     ).send(this)
                 }.onFailure {
                     eventRouter.safeSend(
@@ -133,7 +137,6 @@ fun main(args: Array<String>) {
                 }
             }
         }
-        aliasToService.forEach { (_, service) -> service.start() }
         readiness = true
 
         awaitShutdown(lock, condition)
@@ -143,6 +146,51 @@ fun main(args: Array<String>) {
     }
 }
 
+private fun configureConnectionParameters(
+    configuration: Configuration,
+    eventRouter: MessageRouter<EventBatch>,
+    rootEvent: Event
+) {
+    when {
+        configuration.parameters != null && configuration.sessionAlias == null -> {
+            sendError(
+                eventRouter,
+                "The connection {parameters} configuration requires {sessionAlias}. Please specify the option or use the {sessions} configuration instead",
+                rootEvent
+            )
+        }
+        configuration.sessions != null && configuration.parameters != null ->
+            sendError(
+                eventRouter,
+                "Configuration can't contain both connection version. It must be {sessions} or {parameters}",
+                rootEvent
+            )
+
+        configuration.parameters != null && configuration.sessions == null && configuration.sessionAlias != null -> {
+            (configuration.sessions as ArrayList<ConnParameters>).plusAssign(
+                ConnParameters(
+                    sessionAlias = configuration.sessionAlias,
+                    initialContextFactory = configuration.parameters.initialContextFactory,
+                    factorylookup = configuration.parameters.factorylookup,
+                    sendQueue = configuration.parameters.sendQueue,
+                    receiveQueue = configuration.parameters.receiveQueue
+                )
+            )
+        }
+    }
+}
+
+private fun sendError(eventRouter: MessageRouter<EventBatch>, reason: String, rootEvent: Event) {
+    eventRouter.safeSend(
+        Event.start().endTimestamp()
+            .status(Event.Status.FAILED)
+            .type("Error")
+            .name(reason),
+        rootEvent.id
+    )
+    error(reason)
+}
+
 private fun MessageRouter<EventBatch>.safeSend(event: Event, parentId: String?) {
     runCatching { send(EventBatch.newBuilder().addEvents(event.toProtoEvent(parentId)).build()) }
         .onFailure { LOGGER.error(it) { "Cannot send event with id ${event.id}" } }

src/main/kotlin/com/exactpro/th2/conn/amqp/configuration/Configuration.kt

@@ -20,7 +20,9 @@ class Configuration(
     val enableMessageSendingEvent: Boolean = true,
     val drainIntervalMills: Long = 1000L,
     val rootEventName: String = "ConnAmqp",
-    val sessions: List<ConnParameters>
+    var sessions: MutableList<ConnParameters>?,
+    val sessionAlias: String?,
+    val parameters: SingleConnectionConnParameters?
 )
 
 class ConnParameters(
@@ -30,3 +32,9 @@ class ConnParameters(
     val sendQueue: String,
     val receiveQueue: String
 )
+class SingleConnectionConnParameters(
+    val initialContextFactory: String,
+    val factorylookup: String,
+    val sendQueue: String,
+    val receiveQueue: String
+)