diff --git a/BUILD b/BUILD
index 3f40714..f42177c 100644
--- a/BUILD
+++ b/BUILD
@@ -3,18 +3,18 @@
 gerrit_plugin(
     name = "rabbitmq",
     srcs = glob(["src/main/java/**/*.java"]),
-    resources = glob(["src/main/resources/**/*"]),
     manifest_entries = [
         "Gerrit-PluginName: rabbitmq",
         "Gerrit-Module: com.googlesource.gerrit.plugins.rabbitmq.Module",
         "Implementation-Title: Gerrit rabbitmq plugin",
         "Implementation-URL: https://gerrit-review.googlesource.com/#/admin/projects/plugins/rabbitmq",
     ],
+    resources = glob(["src/main/resources/**/*"]),
     deps = [
         "@amqp_client//jar",
-        "@commons_codec//jar:neverlink",
-        "@commons_io//jar",
-        "@commons_lang//jar:neverlink",
+        "@commons-codec//jar:neverlink",
+        "@commons-io//jar",
+        "@commons-lang//jar:neverlink",
         "@gson//jar:neverlink",
     ],
 )
diff --git a/external_plugin_deps.bzl b/external_plugin_deps.bzl
index 70ed9e6..0af44ef 100644
--- a/external_plugin_deps.bzl
+++ b/external_plugin_deps.bzl
@@ -3,6 +3,6 @@
 def external_plugin_deps():
     maven_jar(
         name = "amqp_client",
-        artifact = "com.rabbitmq:amqp-client:3.5.2",
-        sha1 = "8d10edd29e08f78349bd1da9d18f81c9f8b90567",
+        artifact = "com.rabbitmq:amqp-client:4.1.1",
+        sha1 = "256f6c92c55a8d3cfae8d32e1a15713baedab184",
     )
diff --git a/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/message/GsonProvider.java b/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/message/GsonProvider.java
index fcc4e26..1b7c975 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/message/GsonProvider.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/message/GsonProvider.java
@@ -15,6 +15,8 @@
 package com.googlesource.gerrit.plugins.rabbitmq.message;
 
 import com.google.common.base.Supplier;
+import com.google.gerrit.reviewdb.client.Project;
+import com.google.gerrit.server.events.ProjectNameKeySerializer;
 import com.google.gerrit.server.events.SupplierSerializer;
 import com.google.gson.Gson;
 import com.google.gson.GsonBuilder;
@@ -24,6 +26,9 @@
 
   @Override
   public Gson get() {
-    return new GsonBuilder().registerTypeAdapter(Supplier.class, new SupplierSerializer()).create();
+    return new GsonBuilder()
+        .registerTypeAdapter(Supplier.class, new SupplierSerializer())
+        .registerTypeAdapter(Project.NameKey.class, new ProjectNameKeySerializer())
+        .create();
   }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/message/MessagePublisher.java b/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/message/MessagePublisher.java
index 1651608..37adac7 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/message/MessagePublisher.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/message/MessagePublisher.java
@@ -30,7 +30,6 @@
 import java.util.Timer;
 import java.util.TimerTask;
 import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.TimeUnit;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -45,9 +44,9 @@
   private final Properties properties;
   private final Gson gson;
   private final Timer monitorTimer = new Timer();
-  private boolean available = true;
-  private EventListener eventListener;
   private final LinkedBlockingQueue<Event> queue = new LinkedBlockingQueue<>(MAX_EVENTS);
+  private final Object sessionMon = new Object();
+  private EventListener eventListener;
   private CancelableRunnable publisher;
   private Thread publisherThread;
 
@@ -61,15 +60,22 @@
     this.gson = gson;
     this.eventListener =
         new EventListener() {
+          private int lostEventCount = 0;
+
           @Override
           public void onEvent(Event event) {
-            try {
-              if (!publisherThread.isAlive()) {
-                publisherThread.start();
+            if (!publisherThread.isAlive()) {
+              publisherThread.start();
+            }
+            if (queue.offer(event)) {
+              if (lostEventCount > 0) {
+                LOGGER.warn("Event queue is no longer full, {} events were lost", lostEventCount);
+                lostEventCount = 0;
               }
-              queue.put(event);
-            } catch (InterruptedException e) {
-              LOGGER.warn("Failed to queue event", e);
+            } else {
+              if (lostEventCount++ % 10 == 0) {
+                LOGGER.error("Event queue is full, lost {} event(s)", lostEventCount);
+              }
             }
           }
         };
@@ -82,20 +88,17 @@
           public void run() {
             while (!canceled) {
               try {
-                if (isEnable() && session.isOpen()) {
-                  Event event = queue.poll(200, TimeUnit.MILLISECONDS);
-                  if (event != null) {
-                    if (isEnable() && session.isOpen()) {
-                      publishEvent(event);
-                    } else {
-                      queue.put(event);
-                    }
+                Event event = queue.take();
+                while (!isConnected() && !canceled) {
+                  synchronized (sessionMon) {
+                    sessionMon.wait(1000);
                   }
-                } else {
-                  Thread.sleep(1000);
+                }
+                if (!publishEvent(event) && !queue.offer(event)) {
+                  LOGGER.error("Event lost: {}", gson.toJson(event));
                 }
               } catch (InterruptedException e) {
-                LOGGER.warn("Interupted while taking event", e);
+                LOGGER.warn("Interupted while waiting for event or connection.", e);
               }
             }
           }
@@ -118,22 +121,22 @@
   @Override
   public void start() {
     publisherThread = new Thread(publisher);
+    publisherThread.setName("rabbitmq-publisher");
     publisherThread.start();
-    if (!session.isOpen()) {
-      session.connect();
+    if (!isConnected()) {
+      connect();
       monitorTimer.schedule(
           new TimerTask() {
             @Override
             public void run() {
-              if (!session.isOpen()) {
+              if (!isConnected()) {
                 LOGGER.info("#start: try to reconnect");
-                session.connect();
+                connect();
               }
             }
           },
           MONITOR_FIRSTTIME_DELAY,
           properties.getSection(Monitor.class).interval);
-      available = true;
     }
   }
 
@@ -149,27 +152,6 @@
       }
     }
     session.disconnect();
-    available = false;
-  }
-
-  @Override
-  public void enable() {
-    available = true;
-  }
-
-  @Override
-  public void disable() {
-    available = false;
-  }
-
-  @Override
-  public boolean isEnable() {
-    return available;
-  }
-
-  @Override
-  public Session getSession() {
-    return session;
   }
 
   @Override
@@ -187,7 +169,19 @@
     return this.eventListener;
   }
 
-  private void publishEvent(Event event) {
-    session.publish(gson.toJson(event));
+  private boolean isConnected() {
+    return session != null && session.isOpen();
+  }
+
+  private boolean publishEvent(Event event) {
+    return session.publish(gson.toJson(event));
+  }
+
+  private void connect() {
+    if (!isConnected() && session.connect()) {
+      synchronized (sessionMon) {
+        sessionMon.notifyAll();
+      }
+    }
   }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/message/Publisher.java b/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/message/Publisher.java
index 81998ec..8ef1b3e 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/message/Publisher.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/message/Publisher.java
@@ -2,21 +2,12 @@
 
 import com.google.gerrit.server.events.EventListener;
 import com.googlesource.gerrit.plugins.rabbitmq.config.Properties;
-import com.googlesource.gerrit.plugins.rabbitmq.session.Session;
 
 public interface Publisher {
   void start();
 
   void stop();
 
-  void enable();
-
-  void disable();
-
-  boolean isEnable();
-
-  Session getSession();
-
   Properties getProperties();
 
   String getName();
diff --git a/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/session/Session.java b/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/session/Session.java
index 180aeda..7d0669f 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/session/Session.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/session/Session.java
@@ -16,9 +16,9 @@
 public interface Session {
   boolean isOpen();
 
-  void connect();
+  boolean connect();
 
   void disconnect();
 
-  void publish(String message);
+  boolean publish(String message);
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/session/type/AMQPSession.java b/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/session/type/AMQPSession.java
index d0a872b..3ae3104 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/session/type/AMQPSession.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/session/type/AMQPSession.java
@@ -21,6 +21,7 @@
 import com.googlesource.gerrit.plugins.rabbitmq.config.section.Message;
 import com.googlesource.gerrit.plugins.rabbitmq.config.section.Monitor;
 import com.googlesource.gerrit.plugins.rabbitmq.session.Session;
+import com.rabbitmq.client.AlreadyClosedException;
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
@@ -120,7 +121,7 @@
         ch.addShutdownListener(channelListener);
         failureCount.set(0);
         LOGGER.info(MSG("Channel #{} opened."), ch.getChannelNumber());
-      } catch (IOException ex) {
+      } catch (IOException | AlreadyClosedException ex) {
         LOGGER.error(MSG("Failed to open channel."), ex);
         failureCount.incrementAndGet();
       }
@@ -133,10 +134,10 @@
   }
 
   @Override
-  public void connect() {
+  public boolean connect() {
     if (connection != null && connection.isOpen()) {
       LOGGER.info(MSG("Already connected."));
-      return;
+      return true;
     }
     AMQP amqp = properties.getSection(AMQP.class);
     LOGGER.info(MSG("Connect to {}..."), amqp.uri);
@@ -157,14 +158,16 @@
         connection = factory.newConnection();
         connection.addShutdownListener(connectionListener);
         LOGGER.info(MSG("Connection established."));
+        return true;
       }
     } catch (URISyntaxException ex) {
       LOGGER.error(MSG("URI syntax error: {}"), amqp.uri);
-    } catch (IOException ex) {
+    } catch (IOException | TimeoutException ex) {
       LOGGER.error(MSG("Connection cannot be opened."), ex);
     } catch (KeyManagementException | NoSuchAlgorithmException ex) {
       LOGGER.error(MSG("Security error when opening connection."), ex);
     }
+    return false;
   }
 
   @Override
@@ -194,7 +197,7 @@
   }
 
   @Override
-  public void publish(String messageBody) {
+  public boolean publish(String messageBody) {
     if (channel == null || !channel.isOpen()) {
       channel = getChannel();
     }
@@ -208,9 +211,13 @@
             message.routingKey,
             properties.getAMQProperties().getBasicProperties(),
             messageBody.getBytes(CharEncoding.UTF_8));
+        return true;
       } catch (IOException ex) {
         LOGGER.error(MSG("Error when sending meessage."), ex);
+        return false;
       }
     }
+    LOGGER.error(MSG("Cannot open channel."));
+    return false;
   }
 }