QUEUE-144: Legacy listener source; deliver through #1746 [do not merge]

README.adoc

@@ -1868,6 +1868,7 @@ For comparison see http://kafka.apache.org/documentation.html[Kafka Documentatio
 * https://github.com/OpenHFT/Chronicle-Queue/tree/develop/docs/FAQ.adoc[FAQ] - questions asked by customers
 * https://github.com/OpenHFT/Chronicle-Queue/tree/develop/docs/How_it_works.adoc[How it works] - more depth on how Chronicle Queue is implemented
 * https://github.com/OpenHFT/Chronicle-Queue/tree/develop/docs/utilities.adoc[Utilities] - lists some useful utilities for working with queue files
+* link:docs/context-listeners.adoc[Queue context listeners] - writes roll-level context and handles reader restarts
 
 ==== Online support
 

docs/context-listeners.adoc

@@ -0,0 +1,65 @@
+= Queue Context Listeners
+
+A context listener writes ordinary method-writer records before an appender's first data document
+in each roll cycle. This is useful for snapshots, definitions or checkpoints that readers need
+before processing that cycle's data.
+
+[source,java]
+----
+SingleChronicleQueue queue = builder.build();
+ExcerptAppender appender = queue.acquireAppender();
+appender.contextListener(Events.class, events -> events.context(contextSnapshot));
+Events events = appender.methodWriter(Events.class);
+----
+
+The listener runs under the queue write lock. It must write through the supplied method writer and
+must be configured before the appender is first used. The writer may be retained for normal use.
+The listener is appender-local, is not retried after failure, and remains owned by the caller.
+Use the thread-local appender returned by `acquireAppender()` when its method writer must share an
+appender-local listener. Alternatively, configure the default listener on the queue builder.
+
+A new appender has no record of context written by an earlier appender. It can therefore write the
+context again when it starts in an existing roll cycle. Context records should be complete and
+idempotent.
+
+== Progressive context
+
+An application that already holds a document can write context without configuring a listener.
+The context DTO can keep its last written count in a transient field and report when it needs
+resending:
+
+[source,java]
+----
+final class ContextSnapshot extends SelfDescribingMarshallable implements ProgressiveContext {
+    private String definition;
+    private transient int lastContextCount = -1;
+
+    @Override
+    public boolean needsResending(int contextCount) {
+        if (contextCount <= lastContextCount)
+            return false;
+        lastContextCount = contextCount;
+        return true;
+    }
+}
+
+try (DocumentContext document = events.writingDocument()) {
+    if (contextSnapshot.needsResending(document.contextCount()))
+        events.context(contextSnapshot);
+    events.event(data);
+}
+----
+
+For Queue, the context count is the roll cycle. The context and data calls above are committed in
+one document. This pattern is not supported with double buffering.
+
+== Named tailer restarts
+
+A named tailer persists its next read position. After a restart it may resume in the middle of a
+cycle and will not reread context stored earlier in that cycle. A context listener does not restore
+the reader's in-memory state.
+
+Applications that require this state must either persist it separately or replay the current cycle
+up to the named tailer's saved index with a temporary unnamed tailer. Replay handlers must rebuild
+context without business side effects, and the temporary tailer must not advance the named tailer.
+If the required roll file has already been removed, recovery needs another durable source.

pom.xml

@@ -65,6 +65,8 @@
         <dependency>
             <groupId>net.openhft</groupId>
             <artifactId>chronicle-wire</artifactId>
+            <!-- Remove once the BOM includes the QUEUE-144 Wire API. -->
+            <version>2026.9-SNAPSHOT</version>
         </dependency>
 
         <dependency>

src/main/java/net/openhft/chronicle/queue/ExcerptAppender.java

@@ -76,6 +76,22 @@ default void writeBytes(@NotNull Bytes<?> bytes) {
     default void pretouch() {
     }
 
+    /**
+     * {@inheritDoc}
+     * <p>
+     * For Queue, an output context is a roll cycle. Each appender's listener runs under the queue
+     * write lock before that appender's first data document in a cycle. It is not called for
+     * metadata, explicit-index writes or append-locked queues. A restarted appender can therefore
+     * write context again in an existing cycle. Context listeners are not supported with double
+     * buffering, and the caller retains ownership of the listener.
+     */
+    @NotNull
+    @Override
+    default <T> ExcerptAppender contextListener(@NotNull Class<T> writerType,
+                                                @NotNull MarshallableOut.ContextListener<? super T> listener) {
+        throw new UnsupportedOperationException();
+    }
+
     /**
      * Creates and returns a new writer proxy for the given interface {@code tclass} and the given {@code additional }
      * interfaces.

src/main/java/net/openhft/chronicle/queue/impl/single/ContextListenerState.java

@@ -0,0 +1,227 @@
+/*
+ * Copyright 2013-2025 chronicle.software; SPDX-License-Identifier: Apache-2.0
+ */
+package net.openhft.chronicle.queue.impl.single;
+
+import net.openhft.chronicle.bytes.MethodWriterBuilder;
+import net.openhft.chronicle.wire.BinaryMethodWriterInvocationHandler;
+import net.openhft.chronicle.wire.DocumentContext;
+import net.openhft.chronicle.wire.DocumentContextHolder;
+import net.openhft.chronicle.wire.MarshallableOut;
+import net.openhft.chronicle.wire.VanillaMethodWriterBuilder;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import static java.util.Objects.requireNonNull;
+
+/** Queue configuration, appender state and locked output used by context listeners. */
+final class ContextListenerState extends DocumentContextHolder implements MarshallableOut {
+    static final ContextListenerState UNSET = new ContextListenerState(false);
+    static final ContextListenerState NONE = new ContextListenerState(true);
+
+    @Nullable
+    private final StoreAppender appender;
+    @Nullable
+    private final StoreAppender.StoreAppenderContext context;
+    @Nullable
+    private final Class<?> writerType;
+    @Nullable
+    private final MarshallableOut.ContextListener<?> listener;
+    @Nullable
+    private final Object methodWriter;
+    private boolean started;
+    private boolean notifying;
+    private int lastContextCount = -1;
+    private int nesting;
+
+    private ContextListenerState(boolean started) {
+        this.appender = null;
+        this.context = null;
+        this.writerType = null;
+        this.listener = null;
+        this.methodWriter = null;
+        this.started = started;
+    }
+
+    ContextListenerState(@Nullable Class<?> writerType,
+                         @Nullable MarshallableOut.ContextListener<?> listener) {
+        this.appender = null;
+        this.context = null;
+        this.writerType = writerType;
+        this.listener = listener;
+        this.methodWriter = null;
+    }
+
+    private ContextListenerState(@NotNull StoreAppender appender,
+                                 @NotNull StoreAppender.StoreAppenderContext context,
+                                 @NotNull Class<?> writerType,
+                                 @NotNull MarshallableOut.ContextListener<?> listener) {
+        this.appender = requireNonNull(appender, "appender");
+        this.context = requireNonNull(context, "context");
+        this.writerType = null;
+        this.listener = requireNonNull(listener, "listener");
+        documentContext(context);
+        this.methodWriter = methodWriter(requireNonNull(writerType, "writerType"));
+    }
+
+    ContextListenerState forAppender(@NotNull StoreAppender appender,
+                                     @NotNull StoreAppender.StoreAppenderContext context) {
+        return listener == null
+                ? UNSET
+                : forAppender(appender, context, writerType, listener);
+    }
+
+    static ContextListenerState forAppender(@NotNull StoreAppender appender,
+                                            @NotNull StoreAppender.StoreAppenderContext context,
+                                            @NotNull Class<?> writerType,
+                                            @NotNull MarshallableOut.ContextListener<?> listener) {
+        return new ContextListenerState(
+                appender, context, writerType, listener);
+    }
+
+    boolean started() {
+        return started;
+    }
+
+    void onWriteAttempt() {
+        if (listener == null)
+            return;
+        if (notifying)
+            throw new IllegalStateException("Cannot write to the appender from within a ContextListener; " +
+                    "write through the supplied method writer instead");
+        started = true;
+    }
+
+    boolean beforeDocument(boolean metaData) {
+        if (listener == null || metaData)
+            return false;
+        return notifyIfNeeded();
+    }
+
+    boolean beforeRawDocument() {
+        if (listener == null)
+            return false;
+        appender.resetPositionForContextListener();
+        return notifyIfNeeded();
+    }
+
+    private boolean notifyIfNeeded() {
+        StoreAppender appender = this.appender;
+        int contextCount = appender.cycle();
+        if (contextCount <= lastContextCount)
+            return false;
+        lastContextCount = contextCount;
+
+        notifying = true;
+        try {
+            try {
+                notifyListener();
+            } finally {
+                rollbackIfNotComplete();
+            }
+            return true;
+        } finally {
+            nesting = 0;
+            notifying = false;
+        }
+    }
+
+    @SuppressWarnings({"rawtypes", "unchecked"})
+    private void notifyListener() {
+        ((MarshallableOut.ContextListener) listener).onNewContext(methodWriter);
+    }
+
+    @Override
+    public DocumentContext writingDocument(boolean metaData) {
+        return notifying
+                ? acquireWritingDocument(metaData)
+                : appender.writingDocument(metaData);
+    }
+
+    @Override
+    public DocumentContext acquireWritingDocument(boolean metaData) {
+        if (!notifying)
+            return appender.acquireWritingDocument(metaData);
+
+        StoreAppender.StoreAppenderContext context = this.context;
+        if (nesting > 0 && context.wire() != null && context.isOpen()) {
+            if (!context.chainedElement()) {
+                assert metaData == context.isMetaData();
+                nesting++;
+            }
+            return this;
+        }
+
+        appender.openContextForContextListener(metaData);
+        nesting = 1;
+        return this;
+    }
+
+    @NotNull
+    @Override
+    public <T> MethodWriterBuilder<T> methodWriterBuilder(boolean metaData, @NotNull Class<T> type) {
+        VanillaMethodWriterBuilder<T> builder = new VanillaMethodWriterBuilder<>(type,
+                appender.queue().wireType(),
+                () -> new BinaryMethodWriterInvocationHandler(type, metaData,
+                        () -> ContextListenerState.this));
+        builder.marshallableOut(this);
+        builder.metaData(metaData);
+        return builder;
+    }
+
+    @Override
+    public void rollbackIfNotComplete() {
+        if (!notifying) {
+            appender.rollbackIfNotComplete();
+            return;
+        }
+        StoreAppender.StoreAppenderContext context = this.context;
+        if (nesting == 0 || !context.isOpen())
+            return;
+        context.chainedElement(false);
+        context.rollbackOnClose();
+        nesting = 1;
+        close();
+    }
+
+    @Override
+    public boolean writingIsComplete() {
+        return notifying
+                ? context.writingIsComplete()
+                : appender.writingIsComplete();
+    }
+
+    @Override
+    public void rollbackOnClose() {
+        requireCallback();
+        context.rollbackOnClose();
+    }
+
+    @Override
+    public void close() {
+        requireCallback();
+        if (nesting == 0)
+            throw new IllegalStateException("No ContextListener document is open");
+        StoreAppender.StoreAppenderContext context = this.context;
+        if (context.chainedElement())
+            return;
+        if (nesting > 1) {
+            nesting--;
+            return;
+        }
+        nesting = 0;
+        appender.closeContextForContextListener();
+    }
+
+    @Override
+    public void reset() {
+        requireCallback();
+        context.reset();
+        nesting = 0;
+    }
+
+    private void requireCallback() {
+        if (!notifying)
+            throw new IllegalStateException("ContextListener document is not active");
+    }
+}

src/main/java/net/openhft/chronicle/queue/impl/single/SingleChronicleQueue.java

@@ -128,6 +128,8 @@ public class SingleChronicleQueue extends AbstractCloseable implements RollingCh
     @NotNull
     private final RollCycle rollCycle;
     final AppenderListener appenderListener;
+    @NotNull
+    private final ContextListenerState contextListenerState;
     protected int sourceId;
     private int cycleFileRenamed = -1;
     @NotNull
@@ -186,6 +188,7 @@ protected SingleChronicleQueue(@NotNull final SingleChronicleQueueBuilder builde
             }
             readOnly = builder.readOnly();
             appenderListener = builder.appenderListener();
+            contextListenerState = builder.contextListenerState();
 
             // ReadonlyTableStore is the no-metadata fallback. A SingleTableStore can also be
             // read-only, but still contains the persisted cycle listing that a read-only queue
@@ -638,6 +641,12 @@ protected StoreFileListener storeFileListener() {
         return storeFileListener;
     }
 
+    @NotNull
+    ContextListenerState newContextListenerState(
+            StoreAppender appender, StoreAppender.StoreAppenderContext context) {
+        return contextListenerState.forAppender(appender, context);
+    }
+
     // used by enterprise CQ
     WireStoreSupplier storeSupplier() {
         return storeSupplier;
@@ -1614,7 +1623,7 @@ private StoreSupplier() {
 
         /**
          * Acquires a {@link SingleChronicleQueueStore} for the specified cycle.
-         * If the store doesn't exist and the strategy is {@link CreateStrategy.CREATE}, it will create a new store.
+         * If the store doesn't exist and the strategy is {@code CreateStrategy.CREATE}, it will create a new store.
          *
          * @param cycle          the cycle to acquire the store for
          * @param createStrategy the strategy for creating or reading the store