Fix multiple exporters issue

runtime/binding-http/src/test/java/io/aklivity/zilla/runtime/binding/http/internal/streams/rfc7230/server/EventIT.java

@@ -19,7 +19,6 @@
 import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.junit.rules.RuleChain.outerRule;
 
-import org.junit.Ignore;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.DisableOnDebug;
@@ -51,7 +50,6 @@ public class EventIT
     public final TestRule chain = outerRule(engine).around(k3po).around(timeout);
 
     @Test
-    @Ignore
     @Configuration("server.event.yaml")
     @Specification({
         "${net}/request.with.headers/client",

runtime/binding-http/src/test/java/io/aklivity/zilla/runtime/binding/http/internal/streams/rfc7540/server/EventIT.java

@@ -19,7 +19,6 @@
 import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.junit.rules.RuleChain.outerRule;
 
-import org.junit.Ignore;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.DisableOnDebug;
@@ -51,7 +50,6 @@ public class EventIT
     public final TestRule chain = outerRule(engine).around(k3po).around(timeout);
 
     @Test
-    @Ignore
     @Configuration("server.event.yaml")
     @Specification({
         "${net}/connection.headers/client",

runtime/engine/src/main/java/io/aklivity/zilla/runtime/engine/Engine.java

@@ -68,6 +68,7 @@
 import io.aklivity.zilla.runtime.engine.internal.Info;
 import io.aklivity.zilla.runtime.engine.internal.LabelManager;
 import io.aklivity.zilla.runtime.engine.internal.Tuning;
+import io.aklivity.zilla.runtime.engine.internal.layouts.EventsLayout;
 import io.aklivity.zilla.runtime.engine.internal.registry.EngineManager;
 import io.aklivity.zilla.runtime.engine.internal.registry.EngineWorker;
 import io.aklivity.zilla.runtime.engine.internal.registry.FileWatcherTask;
@@ -523,10 +524,20 @@ public int supplyLabelId(
 
     private final class EventReader implements MessageReader
     {
+        private final EventsLayout.EventAccessor[] accessors;
         private final EventFW eventRO = new EventFW();
         private int minWorkerIndex;
         private long minTimeStamp;
 
+        EventReader()
+        {
+            accessors = new EventsLayout.EventAccessor[workers.size()];
+            for (int i = 0; i < workers.size(); i++)
+            {
+                accessors[i] = workers.get(i).createEventAccessor();
+            }
+        }
+
         @Override
         public int read(
             MessageConsumer handler,
@@ -542,7 +553,7 @@ public int read(
                 for (int j = 0; j < workers.size(); j++)
                 {
                     final int workerIndex = j;
-                    int eventPeeked = workers.get(workerIndex).peekEvent((m, b, i, l) ->
+                    int eventPeeked = accessors[workerIndex].peekEvent((m, b, i, l) ->
                     {
                         eventRO.wrap(b, i, i + l);
                         if (eventRO.timestamp() < minTimeStamp)
@@ -556,7 +567,7 @@ public int read(
                 empty = eventCount == 0;
                 if (!empty)
                 {
-                    messagesRead += workers.get(minWorkerIndex).readEvent(handler, 1);
+                    messagesRead += accessors[minWorkerIndex].readEvent(handler, 1);
                 }
             }
             return messagesRead;

runtime/engine/src/main/java/io/aklivity/zilla/runtime/engine/internal/layouts/EventsLayout.java

@@ -28,7 +28,11 @@
 import java.nio.file.Path;
 import java.nio.file.StandardCopyOption;
 import java.text.SimpleDateFormat;
+import java.util.ArrayList;
 import java.util.Date;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Queue;
 
 import org.agrona.CloseHelper;
 import org.agrona.DirectBuffer;
@@ -46,20 +50,19 @@ public final class EventsLayout implements AutoCloseable
 {
     private final Path path;
     private final long capacity;
+    private final List<EventAccessor> accessors;
 
     private RingBuffer buffer;
-    private RingBufferSpy bufferSpy;
 
     private EventsLayout(
         Path path,
         long capacity,
-        RingBuffer buffer,
-        RingBufferSpy bufferSpy)
+        RingBuffer buffer)
     {
         this.path = path;
         this.capacity = capacity;
         this.buffer = buffer;
-        this.bufferSpy = bufferSpy;
+        this.accessors = new ArrayList<>();
     }
 
     @Override
@@ -82,17 +85,12 @@ public void writeEvent(
         }
     }
 
-    public int readEvent(
-        MessageConsumer handler,
-        int messageCountLimit)
+    public EventAccessor createEventAccessor()
     {
-        return bufferSpy.spy(handler, messageCountLimit);
-    }
-
-    public int peekEvent(
-        MessageConsumer handler)
-    {
-        return bufferSpy.peek(handler);
+        RingBufferSpy ringBufferSpy = createRingBufferSpy();
+        EventAccessor accessor = new EventAccessor(ringBufferSpy);
+        accessors.add(accessor);
+        return accessor;
     }
 
     private void rotateFile()
@@ -110,7 +108,15 @@ private void rotateFile()
             rethrowUnchecked(ex);
         }
         buffer = createRingBuffer(path, capacity);
-        bufferSpy = createRingBufferSpy(path);
+        accessors.forEach(a -> a.addNextBufferSpy(createRingBufferSpy()));
+    }
+
+    private RingBufferSpy createRingBufferSpy()
+    {
+        AtomicBuffer atomicBuffer = createAtomicBuffer(path, 0, false);
+        OneToOneRingBufferSpy spy = new OneToOneRingBufferSpy(atomicBuffer);
+        spy.spyAt(ZERO);
+        return spy;
     }
 
     private static AtomicBuffer createAtomicBuffer(
@@ -135,13 +141,48 @@ private static RingBuffer createRingBuffer(
         return new OneToOneRingBuffer(atomicBuffer);
     }
 
-    private static RingBufferSpy createRingBufferSpy(
-        Path path)
+    public static final class EventAccessor
     {
-        AtomicBuffer atomicBuffer = createAtomicBuffer(path, 0, false);
-        OneToOneRingBufferSpy spy = new OneToOneRingBufferSpy(atomicBuffer);
-        spy.spyAt(ZERO);
-        return spy;
+        private final Queue<RingBufferSpy> nextBufferSpies;
+        private RingBufferSpy bufferSpy;
+
+        private EventAccessor(
+            RingBufferSpy bufferSpy)
+        {
+            this.nextBufferSpies = new LinkedList<>();
+            this.bufferSpy = bufferSpy;
+        }
+
+        public int readEvent(
+            MessageConsumer handler,
+            int messageCountLimit)
+        {
+            int result = bufferSpy.spy(handler, messageCountLimit);
+            if (result == 0 && !nextBufferSpies.isEmpty())
+            {
+                bufferSpy = nextBufferSpies.poll();
+                result = bufferSpy.spy(handler, messageCountLimit);
+            }
+            return result;
+        }
+
+        public int peekEvent(
+            MessageConsumer handler)
+        {
+            int result = bufferSpy.peek(handler);
+            if (result == 0 && !nextBufferSpies.isEmpty())
+            {
+                bufferSpy = nextBufferSpies.poll();
+                result = bufferSpy.peek(handler);
+            }
+            return result;
+        }
+
+        private void addNextBufferSpy(
+            RingBufferSpy bufferSpy)
+        {
+            this.nextBufferSpies.add(bufferSpy);
+        }
     }
 
     public static final class Builder
@@ -166,8 +207,7 @@ public Builder path(
         public EventsLayout build()
         {
             RingBuffer ringBuffer = createRingBuffer(path, capacity);
-            RingBufferSpy ringBufferSpy = createRingBufferSpy(path);
-            return new EventsLayout(path, capacity, ringBuffer, ringBufferSpy);
+            return new EventsLayout(path, capacity, ringBuffer);
         }
     }
 }

runtime/engine/src/main/java/io/aklivity/zilla/runtime/engine/internal/registry/EngineWorker.java

@@ -225,7 +225,7 @@ public class EngineWorker implements EngineContext, Agent
     private final HistogramsLayout histogramsLayout;
     private final EventsLayout eventsLayout;
     private final Supplier<MessageReader> supplyEventReader;
-    private final EventFormatter eventFormatter;
+    private final EventFormatterFactory eventFormatterFactory;
 
     private long initialId;
     private long promiseId;
@@ -435,7 +435,7 @@ public EngineWorker(
         this.errorHandler = errorHandler;
         this.exportersById = new Long2ObjectHashMap<>();
         this.supplyEventReader = supplyEventReader;
-        this.eventFormatter = eventFormatterFactory.create(config, this);
+        this.eventFormatterFactory = eventFormatterFactory;
     }
 
     public static int indexOfId(
@@ -1718,17 +1718,9 @@ public MessageConsumer supplyReceiver(
         return writersByIndex.computeIfAbsent(remoteIndex, supplyWriter);
     }
 
-    public int readEvent(
-        MessageConsumer handler,
-        int messageCountLimit)
+    public EventsLayout.EventAccessor createEventAccessor()
     {
-        return eventsLayout.readEvent(handler, messageCountLimit);
-    }
-
-    public int peekEvent(
-        MessageConsumer handler)
-    {
-        return eventsLayout.peekEvent(handler);
+        return eventsLayout.createEventAccessor();
     }
 
     public MessageReader supplyEventReader()
@@ -1738,7 +1730,7 @@ public MessageReader supplyEventReader()
 
     public EventFormatter supplyEventFormatter()
     {
-        return this.eventFormatter;
+        return eventFormatterFactory.create(config, this);
     }
 
     private MessageConsumer supplyWriter(

runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/internal/layouts/EventsLayoutTest.java

@@ -15,8 +15,8 @@
  */
 package io.aklivity.zilla.runtime.engine.internal.layouts;
 
+import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.equalTo;
-import static org.junit.Assert.assertThat;
 
 import java.nio.file.Path;
 import java.nio.file.Paths;
@@ -42,9 +42,10 @@ public void shouldWriteAndReadEvents()
             .build();
         layout.writeEvent(42, new UnsafeBuffer(), 0, 0);
         msgTypeId = 0;
+        EventsLayout.EventAccessor accessor = layout.createEventAccessor();
 
         // WHEN
-        int count = layout.readEvent(this::readEvent, 1);
+        int count = accessor.readEvent(this::readEvent, 1);
 
         // THEN
         assertThat(count, equalTo(1));