Allow parser sub-classes to redefine the callback mananger

common/src/main/java/com/dmdirc/parser/common/BaseParser.java

@@ -72,7 +72,7 @@ public abstract class BaseParser extends ThreadedParser {
     private URI proxy;
 
     /** The callback manager to use for this parser. */
-    private final CallbackManager callbackManager;
+    private CallbackManager callbackManager;
 
     /**
      * Creates a new base parser for the specified URI.
@@ -90,7 +90,7 @@ public abstract class BaseParser extends ThreadedParser {
             "CallToPrintStackTrace",
             "UseOfSystemOutOrSystemErr"
     })
-    private void handleCallbackError(final PublicationError e) {
+    protected void handleCallbackError(final PublicationError e) {
         if (Thread.holdsLock(errorHandlerLock)) {
             // ABORT ABORT ABORT - we're publishing an error on the same thread we just tried
             // to publish an error on. Something in the error reporting pipeline must be
@@ -175,6 +175,23 @@ public abstract class BaseParser extends ThreadedParser {
         return callbackManager;
     }
 
+    /**
+     * Change the {@link CallbackManager} in use by this parser.
+     *
+     * This will shutdown the old CallbackManager before setting the new one.
+     *
+     * Subscribers are not carried across between callback managers.
+     *
+     * @param manager new CallbackManager to use (ignored if null)
+     */
+    protected void setCallbackManager(final CallbackManager manager) {
+        if (manager == null) {
+            return;
+        }
+        callbackManager.shutdown();
+        callbackManager = manager;
+    }
+
     @Override
     public void joinChannel(final String channel) {
         joinChannels(new ChannelJoinRequest(channel));

common/src/main/java/com/dmdirc/parser/common/CallbackManager.java

@@ -36,11 +36,15 @@ import net.engio.mbassy.bus.error.IPublicationErrorHandler;
 public class CallbackManager extends MBassador<ParserEvent> {
 
     public CallbackManager(final IPublicationErrorHandler errorHandler) {
-        super(new BusConfiguration().addFeature(Feature.SyncPubSub.Default())
+        this(new BusConfiguration().addFeature(Feature.SyncPubSub.Default())
                 .addFeature(Feature.AsynchronousHandlerInvocation.Default(1, 1))
                 .addFeature(Feature.AsynchronousMessageDispatch.Default()
                         .setNumberOfMessageDispatchers(1))
                 .addPublicationErrorHandler(errorHandler));
     }
 
+    protected CallbackManager(final BusConfiguration busConfiguration) {
+        super(busConfiguration);
+    }
+
 }

irc/src/main/java/com/dmdirc/parser/irc/IRCParser.java

@@ -274,6 +274,7 @@ public class IRCParser extends BaseSocketAwareParser implements SecureParser, En
      */
     public IRCParser(final MyInfo myDetails, final URI uri) {
         super(uri);
+        setCallbackManager(new IRCParserCallbackManager(this::handleCallbackError));
 
         // TODO: There should be a factory or builder for parsers that can construct the graph
         final ObjectGraph graph = ObjectGraph.create(new IRCParserModule(this, prefixModes,

irc/src/main/java/com/dmdirc/parser/irc/IRCParserCallbackManager.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright (c) 2006-2017 DMDirc Developers
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package com.dmdirc.parser.irc;
+
+import com.dmdirc.parser.common.CallbackManager;
+import com.dmdirc.parser.events.ParserEvent;
+import net.engio.mbassy.bus.IMessagePublication;
+import net.engio.mbassy.bus.config.BusConfiguration;
+import net.engio.mbassy.bus.config.Feature;
+import net.engio.mbassy.bus.error.IPublicationErrorHandler;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Parser Callback Manager.
+ * Manages adding/removing/calling callbacks.
+ */
+public class IRCParserCallbackManager extends CallbackManager {
+    public IRCParserCallbackManager(final IPublicationErrorHandler errorHandler) {
+        super(new BusConfiguration().addFeature(Feature.SyncPubSub.Default())
+                .addFeature(Feature.AsynchronousHandlerInvocation.Default(1, 1))
+                .addFeature(Feature.AsynchronousMessageDispatch.Default()
+                        .setNumberOfMessageDispatchers(0))
+                .addPublicationErrorHandler(errorHandler));
+    }
+
+    @Override
+    public IMessagePublication publishAsync(final ParserEvent message) {
+        throw new UnsupportedOperationException("IRCParser does not support publishAsync");
+    }
+
+    @Override
+    public IMessagePublication publishAsync(final ParserEvent message, final long timeout, final TimeUnit unit) {
+        throw new UnsupportedOperationException("IRCParser does not support publishAsync");
+    }
+
+}