package io.aeron.cluster;

import io.aeron.Aeron;
import io.aeron.CommonContext;
import io.aeron.cluster.client.AeronCluster;
import io.aeron.cluster.service.ClientSession;
import io.aeron.cluster.service.ClusteredServiceContainer;
import io.aeron.driver.MediaDriver;
import io.aeron.driver.ThreadingMode;
import io.aeron.logbuffer.Header;
import org.agrona.CloseHelper;
import org.agrona.DirectBuffer;
import org.agrona.ExpandableArrayBuffer;
import org.junit.Test;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

public class MultipleClusteredServicesTest
{
    private final AtomicInteger serviceAMessageCount = new AtomicInteger(0);
    private final AtomicInteger serviceBMessageCount = new AtomicInteger(0);

    private final class ServiceA extends TestNode.TestService
    {
        public void onSessionMessage(
            final ClientSession session,
            final long timestamp,
            final DirectBuffer buffer,
            final int offset,
            final int length,
            final Header header)
        {
            serviceAMessageCount.incrementAndGet();
        }
    }

    private final class ServiceB extends TestNode.TestService
    {
        public void onSessionMessage(
            final ClientSession session,
            final long timestamp,
            final DirectBuffer buffer,
            final int offset,
            final int length,
            final Header header)
        {
            serviceBMessageCount.incrementAndGet();
        }
    }

    @Test(timeout = 10_000L)
    public void testMultiService()
    {
        final List<TestCluster.NodeContext> nodeContexts = new ArrayList<>();
        final List<TestCluster.ServiceContext> serviceContexts = new ArrayList<>();
        final List<ClusteredMediaDriver> clusteredMediaDrivers = new ArrayList<>();
        final List<ClusteredServiceContainer> clusteredServiceContainers = new ArrayList<>();

        nodeContexts.add(TestCluster.nodeContext(0, true));
        nodeContexts.add(TestCluster.nodeContext(1, true));
        nodeContexts.add(TestCluster.nodeContext(2, true));

        serviceContexts.add(TestCluster.serviceContext(0, 0, nodeContexts.get(0), ServiceA::new));
        serviceContexts.add(TestCluster.serviceContext(0, 1, nodeContexts.get(0), ServiceB::new));
        serviceContexts.add(TestCluster.serviceContext(1, 0, nodeContexts.get(1), ServiceA::new));
        serviceContexts.add(TestCluster.serviceContext(1, 1, nodeContexts.get(1), ServiceB::new));
        serviceContexts.add(TestCluster.serviceContext(2, 0, nodeContexts.get(2), ServiceA::new));
        serviceContexts.add(TestCluster.serviceContext(2, 1, nodeContexts.get(2), ServiceB::new));

        nodeContexts.forEach(context -> clusteredMediaDrivers.add(ClusteredMediaDriver.launch(
            context.mediaDriverContext, context.archiveContext, context.consensusModuleContext)));

        serviceContexts.forEach(context ->
        {
            final Aeron aeron = Aeron.connect(context.aeron);
            context.aeronArchiveContext.aeron(aeron).ownsAeronClient(false);
            clusteredServiceContainers.add(
                ClusteredServiceContainer.launch(context.serviceContainerContext.aeron(aeron).ownsAeronClient(true)));
        });

        final String aeronDirName = CommonContext.getAeronDirectoryName();

        final MediaDriver clientMediaDriver = MediaDriver.launch(new MediaDriver.Context()
            .threadingMode(ThreadingMode.SHARED)
            .dirDeleteOnStart(true)
            .dirDeleteOnShutdown(true)
            .aeronDirectoryName(aeronDirName));

        final AeronCluster client = AeronCluster.connect(new AeronCluster.Context()
            .aeronDirectoryName(aeronDirName)
            .ingressChannel("aeron:udp")
            .clusterMemberEndpoints(TestCluster.clientMemberEndpoints(3)));

        try
        {
            while (client.offer(new ExpandableArrayBuffer(100), 0, 100) < 0)
            {
                Thread.yield();
                TestUtil.checkInterruptedStatus();
            }

            // Comment out the while loop to see more failures.
            while (serviceAMessageCount.get() < 3)
            {
                Thread.yield();
                TestUtil.checkInterruptedStatus();
            }

            while (serviceBMessageCount.get() < 3)
            {
                Thread.yield();
                TestUtil.checkInterruptedStatus();
            }
        }
        finally
        {
            CloseHelper.closeAll(client, clientMediaDriver);
            clusteredServiceContainers.forEach(CloseHelper::close);
            clusteredMediaDrivers.forEach(CloseHelper::close);
        }
    }
}

@@ -1,8 +1,609 @@
+/*%0A * Copyright 2014-2019 Real Logic Ltd.%0A *%0A * Licensed under the Apache License, Version 2.0 (the %22License%22);%0A * you may not use this file except in compliance with the License.%0A * You may obtain a copy of the License at%0A *%0A * https://www.apache.org/licenses/LICENSE-2.0%0A *%0A * Unless required by applicable law or agreed to in writing, software%0A * distributed under the License is distributed on an %22AS IS%22 BASIS,%0A * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.%0A * See the License for the specific language governing permissions and%0A * limitations under the License.%0A */%0A
 package 
@@ -1209,32 +1209,24 @@
 sTest%0A%7B%0A    
-private 
 final Atomic
@@ -1278,32 +1278,24 @@
 ger(0);%0A    
-private 
 final Atomic
@@ -1348,32 +1348,24 @@
 er(0);%0A%0A    
-private 
 final class 
@@ -1740,16 +1740,8 @@
     
-private 
 fina
@@ -3332,23 +3332,25 @@
 forEach(
+(
 context
+)
  -%3E clus
@@ -3532,23 +3532,38 @@
 forEach(
+%0A            (
 context
+)
  -%3E%0A    
@@ -3558,32 +3558,36 @@
 ext) -%3E%0A        
+    
 %7B%0A            fi
@@ -3576,32 +3576,36 @@
   %7B%0A            
+ 
+   
 final Aeron aero
@@ -3638,16 +3638,20 @@
 aeron);%0A
+    
         
@@ -3727,34 +3727,37 @@
             
-clusteredS
+    context.s
 erviceContai
@@ -3763,31 +3763,99 @@
 iner
-s.add(%0A                
+Context.aeron(aeron).ownsAeronClient(true);%0A                clusteredServiceContainers.add(
 Clus
@@ -3918,47 +3918,16 @@
 text
-.aeron(aeron).ownsAeronClient(true)
 ));%0A
+    
     

Tidy up of PR #785.