Improve reconnection logic

When more than one channel adaptor shares the same in/out queue, use the ChannelAdaptor's hashCode instead of a simple Boolean.TRUE indicator to wake the Sender.
jpos · Feb 28, 2025 · 9fc517b · 9fc517b
1 parent 4ba7639
commit 9fc517b
Showing 1 changed file with 18 additions and 6 deletions.
diff --git a/jpos/src/main/java/org/jpos/q2/iso/ChannelAdaptor.java b/jpos/src/main/java/org/jpos/q2/iso/ChannelAdaptor.java
@@ -310,10 +310,22 @@ public void run () {
                     if (!running())
                         break;
                     Object o = sp.in (in, delay);
-                    if (o instanceof ISOMsg m) {
-                        channel.send (m);
-                        msgOutCounter.increment();
+                    if (o instanceof ISOMsg) {
+                        if (!channel.isConnected()) {
+                            // push back the message so it can be handled by another channel adaptor
+                            sp.push(in, o);
+                            continue;
+                        }
+                        channel.send ((ISOMsg) o);
                         tx++;
+                    } else if (o instanceof Integer) {
+                        if ((int)o != hashCode()) {
+                            // STOP indicator seems to be for another channel adaptor
+                            // sharing the same queue push it back and allow the companion
+                            // channel to get it
+                            sp.push (in, o, 500L);
+                            ISOUtil.sleep (1000L); // larger sleep so that the indicator has time to timeout
+                        }
                     }
                     else if (keepAlive && channel.isConnected() && channel instanceof BaseChannel) {
                         ((BaseChannel)channel).sendKeepAlive();
@@ -381,7 +393,7 @@ public void run () {
                         if (!ignoreISOExceptions) {
                             sp.out (reconnect, Boolean.TRUE, delay);
                             disconnect ();
-                            sp.out (in, Boolean.TRUE); // wake-up Sender
+                            sp.push (in, hashCode()); // wake-up Sender
                         }
                         ISOUtil.sleep(1000);
                     }
@@ -390,15 +402,15 @@ public void run () {
                         // getLog().warn ("channel-receiver-"+out, "Read timeout / EOF - reconnecting");
                         sp.out (reconnect, Boolean.TRUE, delay);
                         disconnect ();
-                        sp.out (in, Boolean.TRUE); // wake-up Sender
+                        sp.push (in, hashCode()); // wake-up Sender
                         ISOUtil.sleep(1000);
                     }
                 } catch (Exception e) { 
                     if (running()) {
                         // getLog().warn ("channel-receiver-"+out, e);
                         sp.out (reconnect, Boolean.TRUE, delay);
                         disconnect ();
-                        sp.out (in, Boolean.TRUE); // wake-up Sender
+                        sp.push (in, hashCode()); // wake-up Sender
                         ISOUtil.sleep(1000);
                     }
                 }