content: /*
   * Copyright 2014-2025 Real Logic Limited.
   * Copyright 2015 Kaazing Corporation
   *
   * 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
   *
   * https://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 io.aeron.samples;
  
  import io.aeron.Aeron;
  import io.aeron.FragmentAssembler;
  import io.aeron.Image;
  import io.aeron.Subscription;
  import io.aeron.logbuffer.FragmentHandler;
  import org.agrona.concurrent.BackoffIdleStrategy;
  import org.agrona.concurrent.IdleStrategy;
  import org.agrona.concurrent.ShutdownSignalBarrier;
  
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.atomic.AtomicBoolean;
  
  /**
   * A subscriber application with two subscriptions which can receive fragmented messages.
   * <p>
   * Creates two subscriptions on a given channel subscribed to two different stream IDs.
   * The default STREAM_ID and CHANNEL are configured in {@link SampleConfiguration}. The default
   * channel and stream IDs can be changed by setting Java system properties at the command line, e.g.:
   * {@code -Daeron.sample.channel=aeron:udp?endpoint=localhost:5555 -Daeron.sample.streamId=20}
   */
  public class MultipleSubscribersWithFragmentAssembly
  {
      private static final int FRAGMENT_COUNT_LIMIT = SampleConfiguration.FRAGMENT_COUNT_LIMIT;
      private static final int STREAM_ID_1 = SampleConfiguration.STREAM_ID;
      private static final int STREAM_ID_2 = SampleConfiguration.STREAM_ID + 1;
  
      private static final String CHANNEL = SampleConfiguration.CHANNEL;
  
      /**
       * Main method for launching the process.
       *
       * @param args passed to the process.
       */
      @SuppressWarnings("try")
      public static void main(final String[] args)
      {
          System.out.format("Subscribing to %s on stream ID %d and stream ID %d%n",
              CHANNEL, STREAM_ID_1, STREAM_ID_2);
  
          final Aeron.Context ctx = new Aeron.Context()
              .availableImageHandler(MultipleSubscribersWithFragmentAssembly::eventAvailableImage)
              .unavailableImageHandler(MultipleSubscribersWithFragmentAssembly::eventUnavailableImage);
  
          final FragmentAssembler assembler1 = new FragmentAssembler(reassembledMessage1(STREAM_ID_1));
          final FragmentAssembler assembler2 = new FragmentAssembler(reassembledMessage2(STREAM_ID_2));
  
          final AtomicBoolean running = new AtomicBoolean(true);
          try (ShutdownSignalBarrier ignore = new ShutdownSignalBarrier(() -> running.set(false));
              Aeron aeron = Aeron.connect(ctx);
              Subscription subscription1 = aeron.addSubscription(CHANNEL, STREAM_ID_1);
              Subscription subscription2 = aeron.addSubscription(CHANNEL, STREAM_ID_2))
          {
              final IdleStrategy idleStrategy = new BackoffIdleStrategy(
                  100, 10, TimeUnit.MICROSECONDS.toNanos(1), TimeUnit.MICROSECONDS.toNanos(100));
  
              int idleCount = 0;
  
              while (running.get())
              {
                  final int fragmentsRead1 = subscription1.poll(assembler1, FRAGMENT_COUNT_LIMIT);
                  final int fragmentsRead2 = subscription2.poll(assembler2, FRAGMENT_COUNT_LIMIT);
  
                  if ((fragmentsRead1 + fragmentsRead2) == 0)
                  {
                      idleStrategy.idle(idleCount++);
                  }
                  else
                  {
                      idleCount = 0;
                  }
              }
  
              System.out.println("Shutting down...");
          }
      }
  
      /**
       * Print the information for an available image to stdout.
       *
       * @param image that has been created.
       */
      public static void eventAvailableImage(final Image image)
      {
          final Subscription subscription = image.subscription();
          System.out.format(
              "new image on %s streamId %x sessionId %x from %s%n",
              subscription.channel(), subscription.streamId(), image.sessionId(), image.sourceIdentity());
      }
  
      /**
       * This handler is called when image is unavailable.
       *
       * @param image that has gone inactive.
       */
      public static void eventUnavailableImage(final Image image)
      {
          final Subscription subscription = image.subscription();
          System.out.format(
              "inactive image on %s streamId %d sessionId %x%n",
              subscription.channel(), subscription.streamId(), image.sessionId());
      }
  
      /**
       * Return a reusable, parameterized {@link FragmentHandler} that prints to stdout for the first stream(STREAM).
       *
       * @param streamId to show when printing.
       * @return subscription data handler function that prints the message contents.
       */
      public static FragmentHandler reassembledMessage1(final int streamId)
      {
          return (buffer, offset, length, header) ->
          {
              System.out.format(
                  "message to stream %d from session %x term id %x term offset %d (%d@%d)%n",
                  streamId, header.sessionId(), header.termId(), header.termOffset(), length, offset);
  
              if (length != 10000)
              {
                  System.out.format(
                      "Received message was not assembled properly;" +
                      " received length was %d, but was expecting 10000%n",
                      length);
              }
          };
      }
  
      /**
       * A reusable, parameterised {@link FragmentHandler} that prints to stdout for the second stream (STREAM + 1).
       *
       * @param streamId to show when printing.
       * @return subscription data handler function that prints the message contents.
       */
      public static FragmentHandler reassembledMessage2(final int streamId)
      {
          return (buffer, offset, length, header) ->
          {
              System.out.format(
                  "message to stream %d from session %x term id %x term offset %d (%d@%d)%n",
                  streamId, header.sessionId(), header.termId(), header.termOffset(), length, offset);
  
              if (length != 9000)
              {
                  System.out.format(
                      "Received message was not assembled properly; received length was %d, but was expecting 9000%n",
                      length);
              }
          };
      }
  }
  