From 8fd1e2515449912e4594e09184357df67048b274 Mon Sep 17 00:00:00 2001
From: Henning Baldersheim <balder@yahoo-inc.com>
Date: Fri, 24 Sep 2021 09:46:53 +0200
Subject: Stopping is server is a one way street. Returning BUSY as a temporary
 error will cause retries, instead of fast-fail.

---
 .../com/yahoo/messagebus/jdisc/MbusClient.java     |  6 ++++-
 .../com/yahoo/messagebus/jdisc/MbusServer.java     | 28 +++++++++++++++-------
 2 files changed, 25 insertions(+), 9 deletions(-)

(limited to 'container-messagebus/src/main/java/com/yahoo/messagebus/jdisc')

diff --git a/container-messagebus/src/main/java/com/yahoo/messagebus/jdisc/MbusClient.java b/container-messagebus/src/main/java/com/yahoo/messagebus/jdisc/MbusClient.java
index 922e4140868..009d1619503 100644
--- a/container-messagebus/src/main/java/com/yahoo/messagebus/jdisc/MbusClient.java
+++ b/container-messagebus/src/main/java/com/yahoo/messagebus/jdisc/MbusClient.java
@@ -9,6 +9,8 @@ import com.yahoo.jdisc.handler.ContentChannel;
 import com.yahoo.jdisc.handler.RequestDeniedException;
 import com.yahoo.jdisc.handler.ResponseHandler;
 import com.yahoo.jdisc.service.ClientProvider;
+
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.logging.Level;
 import com.yahoo.messagebus.EmptyReply;
 import com.yahoo.messagebus.Error;
@@ -29,9 +31,10 @@ import java.util.logging.Logger;
 public final class MbusClient extends AbstractResource implements ClientProvider, ReplyHandler {
 
     private static final Logger log = Logger.getLogger(MbusClient.class.getName());
+    private static final AtomicInteger threadId = new AtomicInteger(0);
     private final BlockingQueue<MbusRequest> queue = new LinkedBlockingQueue<>();
     private final ClientSession session;
-    private final Thread thread = new Thread(new SenderTask(), "MbusClient");
+    private final Thread thread;
     private volatile boolean done = false;
     private final ResourceReference sessionReference;
 
@@ -39,6 +42,7 @@ public final class MbusClient extends AbstractResource implements ClientProvider
     public MbusClient(ClientSession session) {
         this.session = session;
         this.sessionReference = session.refer();
+        thread = new Thread(new SenderTask(), "mbus-client-" + threadId.getAndIncrement());
     }
 
     @Override
diff --git a/container-messagebus/src/main/java/com/yahoo/messagebus/jdisc/MbusServer.java b/container-messagebus/src/main/java/com/yahoo/messagebus/jdisc/MbusServer.java
index 67badddddd2..f8e50845bc8 100644
--- a/container-messagebus/src/main/java/com/yahoo/messagebus/jdisc/MbusServer.java
+++ b/container-messagebus/src/main/java/com/yahoo/messagebus/jdisc/MbusServer.java
@@ -10,13 +10,19 @@ import com.yahoo.jdisc.handler.ContentChannel;
 import com.yahoo.jdisc.handler.ResponseHandler;
 import com.yahoo.jdisc.service.CurrentContainer;
 import com.yahoo.jdisc.service.ServerProvider;
+
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.logging.Level;
-import com.yahoo.messagebus.*;
+
+import com.yahoo.messagebus.EmptyReply;
 import com.yahoo.messagebus.Error;
+import com.yahoo.messagebus.ErrorCode;
+import com.yahoo.messagebus.Message;
+import com.yahoo.messagebus.MessageHandler;
+import com.yahoo.messagebus.Reply;
 import com.yahoo.messagebus.shared.ServerSession;
 
 import java.net.URI;
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.logging.Logger;
 
 /**
@@ -24,8 +30,9 @@ import java.util.logging.Logger;
  */
 public final class MbusServer extends AbstractResource implements ServerProvider, MessageHandler {
 
+    private enum State {INITIALIZING, RUNNING, STOPPED}
     private final static Logger log = Logger.getLogger(MbusServer.class.getName());
-    private final AtomicBoolean running = new AtomicBoolean(false);
+    private final AtomicReference<State> runState = new AtomicReference<>(State.INITIALIZING);
     private final CurrentContainer container;
     private final ServerSession session;
     private final URI uri;
@@ -44,26 +51,31 @@ public final class MbusServer extends AbstractResource implements ServerProvider
     public void start() {
         log.log(Level.FINE, "Starting message bus server.");
         session.connect();
-        running.set(true);
+        runState.set(State.RUNNING);
     }
 
     @Override
     public void close() {
         log.log(Level.FINE, "Closing message bus server.");
-        running.set(false);
+        runState.set(State.STOPPED);
     }
 
     @Override
     protected void destroy() {
         log.log(Level.FINE, "Destroying message bus server.");
-        running.set(false);
+        runState.set(State.STOPPED);
         sessionReference.close();
     }
 
     @Override
     public void handleMessage(Message msg) {
-        if (!running.get()) {
-            dispatchErrorReply(msg, ErrorCode.SESSION_BUSY, "Session temporarily closed.");
+        State state = runState.get();
+        if (state == State.INITIALIZING) {
+            dispatchErrorReply(msg, ErrorCode.SESSION_BUSY, "MBusServer not started.");
+            return;
+        }
+        if (state == State.STOPPED) {
+            dispatchErrorReply(msg, ErrorCode.SEND_QUEUE_CLOSED, "MBusServer has been closed.");
             return;
         }
         if (msg.getTrace().shouldTrace(6)) {
-- 
cgit v1.2.3