/*
 * Copyright 2014 - 2015 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;

import uk.co.real_logic.agrona.collections.Int2ObjectHashMap;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;
import java.util.function.Function;

import static uk.co.real_logic.agrona.collections.CollectionUtil.getOrDefault;

/**
 * Threadsafe for getting to {@link uk.co.real_logic.aeron.Subscription}s by channel and streamId.
 */
class ActiveSubscriptions
{
    private static final Function<String, Int2ObjectHashMap<List<Subscription>>> SUPPLIER = (ignore) -> new Int2ObjectHashMap<>();

    private final Map<String, Int2ObjectHashMap<List<Subscription>>> subscriptionByChannelMap = new HashMap<>();

    public synchronized void forEach(final String channel, final int streamId, final Consumer<Subscription> handler)
    {
        final Int2ObjectHashMap<List<Subscription>> subscriptionByStreamIdMap = subscriptionByChannelMap.get(channel);
        if (null != subscriptionByStreamIdMap)
        {
            final List<Subscription> subscriptions = subscriptionByStreamIdMap.get(streamId);
            if (null != subscriptions)
            {
                subscriptions.forEach(handler);
            }
        }
    }

    public synchronized void forEach(final int streamId, final Consumer<Subscription> handler)
    {
        for (final Int2ObjectHashMap<List<Subscription>> idMap : subscriptionByChannelMap.values())
        {
            final List<Subscription> subscriptions = idMap.get(streamId);
            if (null != subscriptions)
            {
                subscriptions.forEach(handler);
            }
        }
    }

    public synchronized void add(final Subscription subscription)
    {
        final Int2ObjectHashMap<List<Subscription>> subscriptionByStreamIdMap =
            getOrDefault(subscriptionByChannelMap, subscription.channel(), SUPPLIER);

        final List<Subscription> subscriptions =
            subscriptionByStreamIdMap.computeIfAbsent(subscription.streamId(), ignore -> new ArrayList<>());

        subscriptions.add(subscription);
    }

    public synchronized void remove(final Subscription subscription)
    {
        final String channel = subscription.channel();
        final int streamId = subscription.streamId();

        final Int2ObjectHashMap<List<Subscription>> subscriptionByStreamIdMap = subscriptionByChannelMap.get(channel);
        if (null != subscriptionByStreamIdMap)
        {
            final List<Subscription> subscriptions = subscriptionByStreamIdMap.get(streamId);
            if (subscriptions.remove(subscription) && subscriptions.isEmpty())
            {
                subscriptionByStreamIdMap.remove(streamId);
                if (subscriptionByStreamIdMap.isEmpty())
                {
                    subscriptionByStreamIdMap.remove(streamId);
                    if (subscriptionByStreamIdMap.isEmpty())
                    {
                        subscriptionByChannelMap.remove(channel);
                    }
                }
            }
        }
    }
}

@@ -740,212 +740,48 @@
 til.
-HashMap;%0Aimport java.util.List;%0Aimport java.util.Map;%0Aimport java.util.function.Consumer;%0Aimport java.util.function.Function;%0A%0Aimport static uk.co.real_logic.agrona.collections.CollectionUtil.getOrDefault
+List;%0Aimport java.util.function.Consumer
 ;%0A%0A/
@@ -864,28 +864,16 @@
  by 
-channel and 
 streamId
 .%0A *
@@ -868,17 +868,16 @@
 streamId
-.
 %0A */%0Acla
@@ -917,37 +917,13 @@
 ate 
-static final Function%3CString,
+final
  Int
@@ -960,698 +960,63 @@
 on%3E%3E
-%3E SUPPLIER = (ignore) -%3E new Int2ObjectHashMap%3C%3E();%0A%0A    private final Map%3CString, Int2ObjectHashMap%3CList%3CSubscription%3E%3E%3E subscriptionByChannelMap = new HashMap%3C%3E();%0A%0A    public synchronized void forEach(final String channel, final int streamId, final Consumer%3CSubscription%3E handler)%0A    %7B%0A        final Int2ObjectHashMap%3CList%3CSubscription%3E%3E subscriptionByStreamIdMap = subscriptionByChannelMap.get(channel);%0A        if (null != subscriptionByStreamIdMap)%0A        %7B%0A            final List%3CSubscription%3E subscriptions = subscriptionByStreamIdMap.get(streamId);%0A            if (null != subscriptions)%0A            %7B%0A                subscriptions.forEach(handler);%0A            %7D%0A        %7D%0A    %7D
+ subscriptionByStreamIdMap = new Int2ObjectHashMap%3C%3E();
 %0A%0A  
@@ -1126,37 +1126,14 @@
     
-for (
 final 
-Int2ObjectHashMap%3C
 List
@@ -1150,17 +1150,8 @@
 ion%3E
-%3E idMap :
  sub
@@ -1163,95 +1163,33 @@
 tion
-ByChannelMap.values())%0A        %7B%0A            final List%3CSubscription%3E subscriptions = i
+s = subscriptionByStreamI
 dMap
@@ -1196,36 +1196,32 @@
 .get(streamId);%0A
-    
         if (null
@@ -1243,34 +1243,26 @@
 ns)%0A        
-  
-  %7B%0A    
+%7B%0A
             
@@ -1293,30 +1293,16 @@
 ndler);%0A
-            %7D%0A
         
@@ -1386,175 +1386,8 @@
   %7B%0A
-        final Int2ObjectHashMap%3CList%3CSubscription%3E%3E subscriptionByStreamIdMap =%0A            getOrDefault(subscriptionByChannelMap, subscription.channel(), SUPPLIER);%0A%0A
     
@@ -1682,284 +1682,48 @@
 nal 
-String channel = subscription.channel();%0A        final int streamId = subscription.streamId();%0A%0A        final Int2ObjectHashMap%3CList%3CSubscription%3E%3E subscriptionByStreamIdMap = subscriptionByChannelMap.get(channel);%0A        if (null != subscriptionByStreamIdMap)%0A        %7B%0A    
+int streamId = subscription.streamId();%0A
     
@@ -1808,36 +1808,32 @@
 eamId);%0A        
-  
-  
 if (subscription
@@ -1887,34 +1887,26 @@
 ())%0A        
-    %7B%0A    
+%7B%0A
             
@@ -1953,36 +1953,32 @@
 d);%0A            
-  
-  
 if (subscription
@@ -2006,36 +2006,32 @@
 ())%0A            
-    
 %7B%0A              
@@ -2024,36 +2024,32 @@
                 
- 
-   
 subscriptionBySt
@@ -2080,197 +2080,8 @@
 d);%0A
-                    if (subscriptionByStreamIdMap.isEmpty())%0A                    %7B%0A                        subscriptionByChannelMap.remove(channel);%0A                    %7D%0A                %7D%0A
     

simplify ActiveSubscriptions and avoid need for channel.