/*
 * Copyright 2014 Real Logic Ltd.
 *
 * 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 uk.co.real_logic.aeron.examples;

import uk.co.real_logic.aeron.Aeron;
import uk.co.real_logic.aeron.Subscription;
import uk.co.real_logic.aeron.mediadriver.MediaDriver;
import uk.co.real_logic.aeron.util.concurrent.AtomicBuffer;
import uk.co.real_logic.aeron.util.protocol.HeaderFlyweight;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.function.Consumer;

/**
 * Example Aeron subscriber application
 */
public class ExampleSubscriber
{
    public static final String DESTINATION = "udp://localhost:40123";

    public static void main(final String[] args)
    {
        final ExecutorService executor = Executors.newFixedThreadPool(3);
        final Aeron.ClientContext aeronContext = new Aeron.ClientContext().errorHandler(ExampleSubscriber::onError);
        final Subscription.DataHandler messageHandler =
            (buffer, offset, length, sessionId) ->
            {
                final byte[] data = new byte[length];
                buffer.getBytes(offset, data);
                System.out.println("Message " + sessionId + " " + new String(data));
            };


        // make a reusable, parameterized event loop function
        final Consumer<Subscription> loop =
            (subscriber) ->
            {
                try
                {
                    while (true)
                    {
                        subscriber.read();
                        Thread.sleep(1000);
                    }
                }
                catch (final Exception ex)
                {
                    ex.printStackTrace();
                }
            };

        try (final MediaDriver driver = ExampleUtil.createEmbeddedMediaDriver();
             final Aeron aeron = ExampleUtil.createAeron(aeronContext);
             final Subscription subscription1 = aeron.newSubscription(DESTINATION, 30, messageHandler);
             // create a subscriber using the fluent style lambda expression
             final Subscription subscription2 = aeron.newSubscription(DESTINATION, 100, messageHandler))
        {
            // spin off the two subscriber threads
            executor.execute(() -> loop.accept(subscription1));
            executor.execute(() -> loop.accept(subscription2));

            // run aeron conductor thread from here
            aeron.run();
        }
        catch (final Exception ex)
        {
            ex.printStackTrace();
        }

        executor.shutdown();
    }

    public static void onError(final String destination,
                               final long sessionId,
                               final long channelId,
                               final String message,
                               final HeaderFlyweight cause)
    {
        System.err.println(message);
    }

    public static class ExampleDataHandler implements Subscription.DataHandler
    {
        private final long channelId;

        public ExampleDataHandler(final long channelId)
        {
            this.channelId = channelId;
        }

        public long channelId()
        {
            return channelId;
        }

        public void onData(final AtomicBuffer buffer,
                           final int offset,
                           final int length,
                           final long sessionId)
        {
            final byte[] data = new byte[length];
            buffer.getBytes(offset, data);
            System.out.println("Message to channel: " + channelId() + ", from: " + sessionId + ", data (" + length +
                                   "@" + offset + ") <<" + new String(data) + ">>");
        }
    }
}

@@ -1159,16 +1159,111 @@
 :40123%22;
+%0A    public static final int CHANNEL_ID_1 = 30;%0A    public static final int CHANNEL_ID_2 = 100;
 %0A%0A    pu
@@ -1716,32 +1716,33 @@
 (offset, data);%0A
+%0A
                 
@@ -1953,19 +1953,21 @@
 (subscri
-ber
+ption
 ) -%3E%0A   
@@ -2093,24 +2093,49 @@
         
+ final int messagesRead =
  subscri
 ber.read
@@ -2126,19 +2126,21 @@
  subscri
-ber
+ption
 .read();
@@ -2168,27 +2168,172 @@
     
-Thread.sleep(1000);
+if (messagesRead == 0)%0A                        %7B%0A                            Thread.sleep(1000); // Do some other work if nothing is read.%0A                        %7D
 %0A   
@@ -2742,18 +2742,28 @@
 NATION, 
-30
+CHANNEL_ID_1
 , messag
@@ -2777,85 +2777,8 @@
 r);%0A
-             // create a subscriber using the fluent style lambda expression%0A
     
@@ -2856,19 +2856,28 @@
 NATION, 
-100
+CHANNEL_ID_2
 , messag
@@ -2947,16 +2947,51 @@
  threads
+ if you want them to be independent
 %0A       
@@ -3137,16 +3137,23 @@
 n aeron 
+client 
 conducto
@@ -4278,16 +4278,17 @@
  data);%0A
+%0A
         
@@ -4352,33 +4352,95 @@
 () +
- %22, from: %22 + sessionId +
+%0A                               %22, from: %22 + sessionId +%0A                              
  %22, 
@@ -4489,20 +4489,16 @@
         
-    
 %22@%22 + of
@@ -4503,16 +4503,47 @@
 offset +
+%0A                              
  %22) %3C%3C%22 

Tidy up ExampleSubscriber