diff --git a/BUILD b/BUILD
index 441ad42..58248ca 100644
--- a/BUILD
+++ b/BUILD
@@ -13,6 +13,23 @@
     deps = [
         ":events-broker-neverlink",
         "@amqp_client//jar",
+        "@stream_client//jar",
+        "@zstd_jni//jar",
+        "@netty_buffer//jar",
+        "@netty_codec//jar",
+        "@netty_codec_base//jar",
+        "@netty_codec_compression//jar",
+        "@netty_codec_marshalling//jar",
+        "@netty_codec_protobuf//jar",
+        "@netty_common//jar",
+        "@netty_handler//jar",
+        "@netty_resolver//jar",
+        "@netty_transport//jar",
+        "@netty_transport_native_unix_common//jar",
+        "@proton_j//jar",
+        "@lz4_java//jar",
+        "@slf4j_api//jar",
+        "@snappy_java//jar",
         "@commons-codec//jar:neverlink",
         "@commons-io//jar",
         "@commons-lang3//jar:neverlink",
diff --git a/external_plugin_deps.bzl b/external_plugin_deps.bzl
index 2827eee..3cda8de 100644
--- a/external_plugin_deps.bzl
+++ b/external_plugin_deps.bzl
@@ -6,3 +6,106 @@
         artifact = "com.rabbitmq:amqp-client:5.10.0",
         sha1 = "4de351467a13b8ca4eb7e8023032f9f964a21796",
     )
+
+    maven_jar(
+        name = "stream_client",
+        artifact = "com.rabbitmq:stream-client:1.1.0",
+        sha1 = "585736dd2da4d3da6e72adeef4103cdb362435eb",
+    )
+
+    # Transitive dependencies for stream-client below
+    maven_jar(
+        name = "zstd_jni",
+        artifact = "com.github.luben:zstd-jni:1.5.7-3",
+        sha1 = "dc55c256583fb810f9fe3ede271843b68132fb7c",
+    )
+
+    maven_jar(
+        name = "netty_buffer",
+        artifact = "io.netty:netty-buffer:4.2.1.Final",
+        sha1 = "095e48c931eb31614c06adf775a2118f4297fd97",
+    )
+
+    maven_jar(
+        name = "netty_codec",
+        artifact = "io.netty:netty-codec:4.2.1.Final",
+        sha1 = "55421a0c16ccd25692613e86f75b5a3f8db3ea82",
+    )
+
+    maven_jar(
+        name = "netty_codec_base",
+        artifact = "io.netty:netty-codec-base:4.2.1.Final",
+        sha1 = "ee2a21f30549d6fb0ec1da71ac748c385368d807",
+    )
+
+    maven_jar(
+        name = "netty_codec_compression",
+        artifact = "io.netty:netty-codec-compression:4.2.1.Final",
+        sha1 = "b85ef35b94d1b540c021df6bb778cc8dc086588b",
+    )
+
+    maven_jar(
+        name = "netty_codec_marshalling",
+        artifact = "io.netty:netty-codec-marshalling:4.2.1.Final",
+        sha1 = "838fd5dc1ae854879a9e553a0d217abfd5d7fda9",
+    )
+
+    maven_jar(
+        name = "netty_codec_protobuf",
+        artifact = "io.netty:netty-codec-protobuf:4.2.1.Final",
+        sha1 = "086c04147717b63edadb9e2d1bd371f6091e2ba0",
+    )
+
+    maven_jar(
+        name = "netty_common",
+        artifact = "io.netty:netty-common:4.2.1.Final",
+        sha1 = "ed99ce89380bc3e9b297c9a5d87fbb6e669e1cf4",
+    )
+
+    maven_jar(
+        name = "netty_handler",
+        artifact = "io.netty:netty-handler:4.2.1.Final",
+        sha1 = "1673ad3e66acb5bfaa2dab839ee0d6b22225fbee",
+    )
+
+    maven_jar(
+        name = "netty_resolver",
+        artifact = "io.netty:netty-resolver:4.2.1.Final",
+        sha1 = "80a111e04ee696cca0fe928b7a64787551a848e4",
+    )
+
+    maven_jar(
+        name = "netty_transport",
+        artifact = "io.netty:netty-transport:4.2.1.Final",
+        sha1 = "5c9140fffd02fe833a570dfd00c3dced88c72abd",
+    )
+
+    maven_jar(
+        name = "netty_transport_native_unix_common",
+        artifact = "io.netty:netty-transport-native-unix-common:4.2.1.Final",
+        sha1 = "9edcc29edb12cb33890e6de025fbb60cc3846d6d",
+    )
+
+    maven_jar(
+        name = "proton_j",
+        artifact = "org.apache.qpid:proton-j:0.34.1",
+        sha1 = "e0d6c62cef4929db66dd6df55bee699b2274a9cc",
+    )
+
+    maven_jar(
+        name = "lz4_java",
+        artifact = "org.lz4:lz4-java:1.8.0",
+        sha1 = "4b986a99445e49ea5fbf5d149c4b63f6ed6c6780",
+    )
+
+    maven_jar(
+        name = "slf4j_api",
+        artifact = "org.slf4j:slf4j-api:1.7.36",
+        sha1 = "6c62681a2f655b49963a5983b8b0950a6120ae14",
+    )
+
+    maven_jar(
+        name = "snappy_java",
+        artifact = "org.xerial.snappy:snappy-java:1.1.10.7",
+        sha1 = "3049f95640f4625a945cfab85715f603fa4c8f80",
+    )
diff --git a/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/Module.java b/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/Module.java
index 56acd18..6984f98 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/Module.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/Module.java
@@ -41,6 +41,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.config.section.Section;
+import com.googlesource.gerrit.plugins.rabbitmq.config.section.Stream;
 import com.googlesource.gerrit.plugins.rabbitmq.message.BaseProperties;
 import com.googlesource.gerrit.plugins.rabbitmq.message.BasePropertiesProvider;
 import com.googlesource.gerrit.plugins.rabbitmq.message.GerritEventPublisher;
@@ -50,7 +51,7 @@
 import com.googlesource.gerrit.plugins.rabbitmq.session.PublisherSession;
 import com.googlesource.gerrit.plugins.rabbitmq.session.SubscriberSession;
 import com.googlesource.gerrit.plugins.rabbitmq.session.type.AMQPPublisherSession;
-import com.googlesource.gerrit.plugins.rabbitmq.session.type.AMQPSubscriberSession;
+import com.googlesource.gerrit.plugins.rabbitmq.session.type.SubscriberSessionFactoryImpl;
 import com.googlesource.gerrit.plugins.rabbitmq.worker.DefaultEventWorker;
 import com.googlesource.gerrit.plugins.rabbitmq.worker.EventWorker;
 import com.googlesource.gerrit.plugins.rabbitmq.worker.EventWorkerFactory;
@@ -97,6 +98,7 @@
     sectionBinder.addBinding().to(Message.class);
     sectionBinder.addBinding().to(Monitor.class);
     sectionBinder.addBinding().to(General.class);
+    sectionBinder.addBinding().to(Stream.class);
 
     install(
         new FactoryModuleBuilder()
@@ -104,10 +106,6 @@
             .build(AMQPPublisherSession.Factory.class));
     install(
         new FactoryModuleBuilder()
-            .implement(SubscriberSession.class, AMQPSubscriberSession.class)
-            .build(AMQPSubscriberSession.Factory.class));
-    install(
-        new FactoryModuleBuilder()
             .implement(Publisher.class, GerritEventPublisher.class)
             .build(GerritEventPublisherFactory.class));
     install(
@@ -118,6 +116,9 @@
         new FactoryModuleBuilder()
             .implement(EventWorker.class, UserEventWorker.class)
             .build(EventWorkerFactory.class));
+    bind(SubscriberSession.Factory.class)
+        .to(SubscriberSessionFactoryImpl.class)
+        .in(Singleton.class);
     bind(Gson.class).toProvider(GsonProvider.class).in(Singleton.class);
 
     DynamicSet.bind(binder(), LifecycleListener.class).to(Manager.class);
diff --git a/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/RabbitMqBrokerApi.java b/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/RabbitMqBrokerApi.java
index ed72e1c..649ac8a 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/RabbitMqBrokerApi.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/RabbitMqBrokerApi.java
@@ -74,8 +74,11 @@
 
   @Override
   public void replayAllEvents(String topic) {
-    throw new NotImplementedException(
-        "The RabbitMqBrokerApi does not support replayAllEvents yet.");
+    for (TopicSubscriber topicSubscriber : topicSubscribers) {
+      if (topicSubscriber.topic().equals(topic)) {
+        subscribers.replayAllEvents(topicSubscriber);
+      }
+    }
   }
 
   @Override
diff --git a/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/config/section/Stream.java b/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/config/section/Stream.java
new file mode 100644
index 0000000..7ac3059
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/config/section/Stream.java
@@ -0,0 +1,53 @@
+// Copyright (C) 2025 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.googlesource.gerrit.plugins.rabbitmq.config.section;
+
+import com.google.common.flogger.FluentLogger;
+import com.googlesource.gerrit.plugins.rabbitmq.annotation.Default;
+import java.util.ArrayList;
+import java.util.List;
+
+public class Stream implements Section {
+
+  private static final FluentLogger logger = FluentLogger.forEnclosingClass();
+
+  @Default("false")
+  public Boolean enabled;
+
+  @Default public String uri;
+
+  @Default public String streamPrefix;
+
+  @Default public String consumerPrefix;
+
+  @Default("500")
+  public Integer windowSize;
+
+  public boolean isValid() {
+    List<String> missingSettings = new ArrayList<>(3);
+
+    if (uri.isEmpty()) missingSettings.add("uri");
+    if (streamPrefix.isEmpty()) missingSettings.add("streamPrefix");
+    if (consumerPrefix.isEmpty()) missingSettings.add("consumerPrefix");
+
+    if (!missingSettings.isEmpty()) {
+      logger.atSevere().log(
+          "Missing mandatory stream settings: %s, please set these if you want to use streams.",
+          missingSettings);
+      return false;
+    }
+    return true;
+  }
+}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/message/BrokerApiSubscribers.java b/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/message/BrokerApiSubscribers.java
index ed80334..d7acc92 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/message/BrokerApiSubscribers.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/message/BrokerApiSubscribers.java
@@ -30,8 +30,9 @@
 import com.google.inject.Inject;
 import com.google.inject.Singleton;
 import com.googlesource.gerrit.plugins.rabbitmq.config.Properties;
+import com.googlesource.gerrit.plugins.rabbitmq.config.section.Stream;
 import com.googlesource.gerrit.plugins.rabbitmq.session.SubscriberSession;
-import com.googlesource.gerrit.plugins.rabbitmq.session.type.AMQPSubscriberSession;
+import com.googlesource.gerrit.plugins.rabbitmq.session.type.StreamSubscriberSession;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ExecutionException;
@@ -48,7 +49,7 @@
 
   @Inject
   public BrokerApiSubscribers(
-      AMQPSubscriberSession.Factory sessionFactory,
+      SubscriberSession.Factory sessionFactory,
       @EventGson Gson gson,
       @BrokerApiProperties Properties properties) {
     this.properties = properties;
@@ -136,4 +137,17 @@
     }
     return session.removeSubscriber(consumerTag);
   }
+
+  public boolean replayAllEvents(TopicSubscriber topicSubscriber) {
+    if (properties.getSection(Stream.class).enabled) {
+      StreamSubscriberSession streamSession = (StreamSubscriberSession) session;
+      streamSession.resetOffset(consumerTags.get(topicSubscriber), 0);
+      removeSubscriber(topicSubscriber);
+      addSubscriber(topicSubscriber);
+      return true;
+    }
+    logger.atWarning().log(
+        "Only streams support the replay functionality, please enable stream support to use this");
+    return false;
+  }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/session/SubscriberSession.java b/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/session/SubscriberSession.java
index 5fd1577..bd0ff42 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/session/SubscriberSession.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/session/SubscriberSession.java
@@ -14,9 +14,14 @@
 
 package com.googlesource.gerrit.plugins.rabbitmq.session;
 
+import com.googlesource.gerrit.plugins.rabbitmq.config.Properties;
 import java.util.function.Consumer;
 
 public interface SubscriberSession extends Session {
+  public interface Factory {
+    SubscriberSession create(Properties properties);
+  }
+
   String addSubscriber(String topic, Consumer<String> messageBodyConsumer);
 
   boolean removeSubscriber(String consumerTag);
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 dcf12e3..e06810a 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
@@ -33,7 +33,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.commons.lang3.StringUtils;
 
-public abstract class AMQPSession implements Session {
+public class AMQPSession implements Session {
   private static final FluentLogger logger = FluentLogger.forEnclosingClass();
 
   private volatile Connection connection;
diff --git a/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/session/type/AMQPSubscriberSession.java b/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/session/type/AMQPSubscriberSession.java
index d19e023..b32537a 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/session/type/AMQPSubscriberSession.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/session/type/AMQPSubscriberSession.java
@@ -38,10 +38,6 @@
 public final class AMQPSubscriberSession extends AMQPSession implements SubscriberSession {
   private static final FluentLogger logger = FluentLogger.forEnclosingClass();
 
-  public interface Factory {
-    SubscriberSession create(Properties properties);
-  }
-
   private volatile Map<String, Channel> channels = new ConcurrentHashMap<>();
 
   @Inject
@@ -69,7 +65,7 @@
 
     try {
       channel.basicQos(amqp.consumerPrefetch > 0 ? amqp.consumerPrefetch : 0);
-    } catch(IOException ex) {
+    } catch (IOException ex) {
       logger.atSevere().withCause(ex).log("Error when trying to set consumer prefetch");
     }
 
diff --git a/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/session/type/StreamSession.java b/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/session/type/StreamSession.java
new file mode 100644
index 0000000..8cf282a
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/session/type/StreamSession.java
@@ -0,0 +1,78 @@
+// Copyright (C) 2025 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.googlesource.gerrit.plugins.rabbitmq.session.type;
+
+import com.google.common.flogger.FluentLogger;
+import com.googlesource.gerrit.plugins.rabbitmq.config.Properties;
+import com.googlesource.gerrit.plugins.rabbitmq.config.section.AMQP;
+import com.googlesource.gerrit.plugins.rabbitmq.config.section.Gerrit;
+import com.googlesource.gerrit.plugins.rabbitmq.config.section.Stream;
+import com.googlesource.gerrit.plugins.rabbitmq.session.Session;
+import com.rabbitmq.stream.Environment;
+import com.rabbitmq.stream.EnvironmentBuilder;
+import org.apache.commons.lang3.StringUtils;
+
+public abstract class StreamSession implements Session {
+  private static final FluentLogger logger = FluentLogger.forEnclosingClass();
+
+  protected volatile Environment environment;
+
+  protected final Properties properties;
+
+  public StreamSession(Properties properties) {
+    this.properties = properties;
+  }
+
+  @Override
+  public boolean isOpen() {
+    if (environment != null) {
+      logger.atFine().log("Assuming stream session is open since Environment exists");
+      return true;
+    }
+    return false;
+  }
+
+  @Override
+  public synchronized boolean connect() {
+    AMQP amqp = properties.getSection(AMQP.class);
+    Stream stream = properties.getSection(Stream.class);
+
+    if (!stream.isValid()) {
+      return false;
+    }
+    EnvironmentBuilder builder = Environment.builder();
+    builder.uri(stream.uri);
+    if (StringUtils.isNotEmpty(amqp.username)) {
+      builder.username(amqp.username);
+    }
+    Gerrit gerrit = properties.getSection(Gerrit.class);
+    String securePassword = gerrit.getAMQPUserPassword(amqp.username);
+    if (StringUtils.isNotEmpty(securePassword)) {
+      builder.password(securePassword);
+    } else if (StringUtils.isNotEmpty(amqp.password)) {
+      builder.password(amqp.password);
+    }
+
+    environment = builder.build();
+    logger.atInfo().log("Environment built for %s.", stream.uri);
+    return true;
+  }
+
+  @Override
+  public void disconnect() {
+    environment.close();
+    environment = null;
+  }
+}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/session/type/StreamSubscriberSession.java b/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/session/type/StreamSubscriberSession.java
new file mode 100644
index 0000000..a266beb
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/session/type/StreamSubscriberSession.java
@@ -0,0 +1,204 @@
+// Copyright (C) 2025 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.googlesource.gerrit.plugins.rabbitmq.session.type;
+
+import com.google.common.flogger.FluentLogger;
+import com.google.inject.Inject;
+import com.google.inject.assistedinject.Assisted;
+import com.googlesource.gerrit.plugins.rabbitmq.config.Properties;
+import com.googlesource.gerrit.plugins.rabbitmq.config.section.Exchange;
+import com.googlesource.gerrit.plugins.rabbitmq.config.section.Stream;
+import com.googlesource.gerrit.plugins.rabbitmq.session.SubscriberSession;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.stream.Message;
+import com.rabbitmq.stream.MessageHandler;
+import com.rabbitmq.stream.NoOffsetException;
+import com.rabbitmq.stream.OffsetSpecification;
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+
+public final class StreamSubscriberSession extends StreamSession implements SubscriberSession {
+  private static final FluentLogger logger = FluentLogger.forEnclosingClass();
+
+  private volatile Consumers consumers = new Consumers();
+
+  @Inject
+  public StreamSubscriberSession(@Assisted Properties properties) {
+    super(properties);
+  }
+
+  @Override
+  public void disconnect() {
+    logger.atInfo().log("Disconnecting subscriber session...");
+    consumers.close();
+    super.disconnect();
+  }
+
+  @Override
+  public String addSubscriber(String topic, Consumer<String> messageBodyConsumer) {
+    if (environment == null) {
+      if (!connect()) {
+        logger.atSevere().log("Failed to connect to rabbitMQ with environment");
+        return null;
+      }
+    }
+    Stream streamProp = properties.getSection(Stream.class);
+
+    String streamName = streamProp.streamPrefix + "." + topic;
+    String consumerName = streamProp.consumerPrefix + "." + topic;
+    if (!environment.streamExists(streamName)) {
+      environment.streamCreator().stream(streamName).create();
+    }
+
+    bindStreamToExchange(streamName, topic);
+
+    String consumerId = UUID.randomUUID().toString();
+    com.rabbitmq.stream.Consumer consumer =
+        environment.consumerBuilder().stream(streamName)
+            .offset(OffsetSpecification.first())
+            .name(consumerName)
+            .manualTrackingStrategy()
+            .builder()
+            .messageHandler(new Handler(topic, messageBodyConsumer, consumerId))
+            .build();
+    try {
+      logger.atInfo().log(
+          "Consumer added for topic %s, consuming from stream-offset %d",
+          topic, consumer.storedOffset());
+    } catch (NoOffsetException ex) {
+      logger.atInfo().withCause(ex).log(
+          "No offset found for consumer that listens on topic %s, consuming from start of stream.",
+          topic);
+    }
+
+    consumers.put(consumerId, consumer, false);
+    return consumerId;
+  }
+
+  private void bindStreamToExchange(String streamName, String topic) {
+    String exchangeName = properties.getSection(Exchange.class).name;
+    AMQPSession session = new AMQPSession(properties);
+    try (Channel channel = session.createChannel()) {
+      channel.queueBind(streamName, exchangeName, topic);
+    } catch (IOException | TimeoutException ex) {
+      logger.atSevere().withCause(ex).log("Failed to bind stream to exchange or close channel");
+    }
+    session.disconnect();
+  }
+
+  @Override
+  public boolean removeSubscriber(String consumerId) {
+    return consumers.closeConsumer(consumerId);
+  }
+
+  public void resetOffset(String consumerId, long offset) {
+    consumers.reset(consumerId, offset);
+  }
+
+  private class Consumers {
+    private volatile Map<String, ConsumerPair> consumersMap = new ConcurrentHashMap<>();
+
+    void put(String consumerId, com.rabbitmq.stream.Consumer consumer, boolean resetOffset) {
+      consumersMap.put(consumerId, new ConsumerPair(consumer, resetOffset));
+    }
+
+    boolean isReset(String consumerId) {
+      return consumersMap.get(consumerId).resetOffset;
+    }
+
+    void reset(String consumerId, long offset) {
+      com.rabbitmq.stream.Consumer consumer = consumersMap.get(consumerId).consumer;
+      if (consumer != null) {
+        synchronized (consumer) {
+          consumersMap.get(consumerId).resetOffset = true;
+          consumer.store(offset);
+        }
+      }
+    }
+
+    boolean closeConsumer(String consumerId) {
+      ConsumerPair pair = consumersMap.remove(consumerId);
+      if (pair == null) {
+        return false;
+      }
+      pair.consumer.close();
+      return true;
+    }
+
+    void close() {
+      synchronized (consumersMap) {
+        Iterator<Entry<String, ConsumerPair>> it = consumersMap.entrySet().iterator();
+        while (it.hasNext()) {
+          it.next().getValue().consumer.close();
+          it.remove();
+        }
+      }
+    }
+
+    private class ConsumerPair {
+      com.rabbitmq.stream.Consumer consumer;
+      boolean resetOffset;
+
+      ConsumerPair(com.rabbitmq.stream.Consumer consumer, boolean resetOffset) {
+        this.consumer = consumer;
+        this.resetOffset = resetOffset;
+      }
+    }
+  }
+
+  private class Handler implements MessageHandler {
+    private AtomicInteger messageConsumed;
+    private String topic;
+    private Consumer<String> messageBodyConsumer;
+    private String consumerId;
+
+    Handler(String topic, Consumer<String> messageBodyConsumer, String consumerId) {
+      messageConsumed = new AtomicInteger(0);
+      this.topic = topic;
+      this.messageBodyConsumer = messageBodyConsumer;
+      this.consumerId = consumerId;
+    }
+
+    @Override
+    public void handle(MessageHandler.Context context, Message message) {
+      Stream prop = properties.getSection(Stream.class);
+      try {
+        logger.atFiner().log(
+            "Consume message from topic %s with offset %d", topic, context.offset());
+        messageBodyConsumer.accept(new String(message.getBodyAsBinary(), "UTF-8"));
+        if (messageConsumed.incrementAndGet() % prop.windowSize == 0) {
+          com.rabbitmq.stream.Consumer consumer = context.consumer();
+          long newOffset = Math.max(context.offset() - prop.windowSize, 0);
+          logger.atFine().log("Store new offset %d for stream with topic %s", newOffset, topic);
+          synchronized (consumer) {
+            if (!consumers.isReset(consumerId)) {
+              consumer.store(newOffset);
+            }
+          }
+        }
+      } catch (IOException ex) {
+        logger.atSevere().withCause(ex).log(
+            "Error handling stream message with id %d", message.getPublishingId());
+      }
+    }
+  }
+}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/session/type/SubscriberSessionFactoryImpl.java b/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/session/type/SubscriberSessionFactoryImpl.java
new file mode 100644
index 0000000..57de967
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/session/type/SubscriberSessionFactoryImpl.java
@@ -0,0 +1,29 @@
+// Copyright (C) 2025 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.googlesource.gerrit.plugins.rabbitmq.session.type;
+
+import com.googlesource.gerrit.plugins.rabbitmq.config.Properties;
+import com.googlesource.gerrit.plugins.rabbitmq.config.section.Stream;
+import com.googlesource.gerrit.plugins.rabbitmq.session.SubscriberSession;
+
+public class SubscriberSessionFactoryImpl implements SubscriberSession.Factory {
+
+  public SubscriberSession create(Properties properties) {
+    if (properties.getSection(Stream.class).enabled) {
+      return new StreamSubscriberSession(properties);
+    }
+    return new AMQPSubscriberSession(properties);
+  }
+}
diff --git a/src/main/resources/Documentation/config.md b/src/main/resources/Documentation/config.md
index 54cdee8..3b18e98 100644
--- a/src/main/resources/Documentation/config.md
+++ b/src/main/resources/Documentation/config.md
@@ -24,6 +24,12 @@
 other configs with the additions of queuePrefix, durable, exclusive and autoDelete that decides
 queue properties and consumerPrefetch that is a channel property related to consumers. The
 event-broker API use its own publisher that is separate from the previously mentioned publisher.
+You can use streams instead of queues if you enable it with stream.enabled. There is a separate
+set of configuration options for streams.
+Note: If you get certificate host verfication failures the problem may be that the RabbitMQ cluster
+advertise the hostname without the domain. You can verify this by running "rabbitmqctl eval 
+'rabbit_stream:tls_host().'" and it can be solved by setting stream.advertised_tls_host as the
+FQDN. For more info check the RabbitMQ documentation.
 
 Secure.config
 ---------------------
@@ -93,6 +99,30 @@
 * `exchange.name`
     * The name of exchange.
 
+* `stream.enabled`
+    * Make the brokerApi subscribe on streams instead of queues, defaults to false. Only used in
+    broker.config.
+
+* `stream.uri`
+    * The URI of RabbitMQ server's endpoint that should be used to create a stream. This needs to
+    be set to use streams. Only used in broker.config.
+
+* `stream.streamPrefix`
+    * The streams that store the messages of the the subscribed topics will be named
+    prefix + '.' + topic. This needs to be set to use streams. Only used in
+    broker.config.
+
+* `stream.consumerPrefix`
+    * The consumers that consume from the streams will be named prefix + '.' + topic. The name is
+    used by the Rabbitmq stream library to retrieve the correct offset from the server. The prefix
+    needs to be unique for each Gerrit instance. This needs to be set to use streams. Only used in
+    broker.config.
+
+* `stream.windowSize`
+    * How many messages that needs to be consumed until we set a new offset. New offset will be
+    offset of the currently proccessed message subtracted by `windowSize`, defaults to 500. Only
+    used in broker.config.
+
 * `general.publishAllGerritEvents`
     * Will publish gerrit stream events to configured exchange automatically if enabled, defaults
       to true.