/*
 * Copyright 2015 Red Hat, Inc. and/or its affiliates
 * and other contributors as indicated by the @author tags.
 *
 * 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 org.hawkular.inventory.api.observable;

import rx.Observable;
import rx.functions.Action0;
import rx.subjects.PublishSubject;
import rx.subjects.Subject;

import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author Lukas Krejci
 * @since 0.0.1
 */
final class ObservableContext {
    private final Map<Interest<?>, Subject<?, ?>> observables = new ConcurrentHashMap<>();

    public <T> Observable<T> getObservableFor(Interest<T> interest) {
        @SuppressWarnings("unchecked")
        Subject<T, T> sub = (Subject<T, T>) observables.get(interest);

        if (sub == null) {
            sub = PublishSubject.<T>create().toSerialized();
            observables.put(interest, sub);
        }

        SubscriptionTracker tracker = new SubscriptionTracker(() -> observables.remove(interest));

        return sub.doOnSubscribe(tracker.onSubscribe()).doOnUnsubscribe(tracker.onUnsubscribe());
    }

    @SuppressWarnings("unchecked")
    public <T> Iterator<Subject<T, T>> matchingSubjects(Action<T> action, T object) {
        return observables.entrySet().stream().filter((e) -> e.getKey().matches(action, object))
                .map((e) -> (Subject<T, T>) e.getValue()).iterator();
    }

    private static class SubscriptionTracker {

        private final AtomicLong counter = new AtomicLong(0);
        private final Runnable action;

        public SubscriptionTracker(Runnable action) {
            this.action = action;
        }

        public Action0 onSubscribe() {
            return counter::incrementAndGet;
        }

        public Action0 onUnsubscribe() {
            return () -> {
                if (counter.decrementAndGet() == 0) {
                    action.run();
                }
            };
        }
    }
}

@@ -1104,12 +1104,19 @@
 ject
-%3C?, 
+AndWrapper%3C
 ?%3E%3E 
@@ -1239,210 +1239,729 @@
     
-@SuppressWarnings(%22unchecked%22)%0A        Subject%3CT, T%3E sub = (Subject%3CT, T%3E) observables.get(interest);%0A%0A        if (sub == null) %7B%0A            sub = PublishSubject.%3CT%3Ecreate().toSerialized();%0A           
+SubjectAndWrapper%3CT%3E sub = getSubjectAndWrapper(interest, true);%0A        return sub.wrapper;%0A    %7D%0A%0A    public boolean isObserved(Interest%3C?%3E interest) %7B%0A        return observables.containsKey(interest);%0A    %7D%0A%0A    @SuppressWarnings(%22unchecked%22)%0A    public %3CT%3E Iterator%3CSubject%3CT, T%3E%3E matchingSubjects(Action%3CT%3E action, T object) %7B%0A        return observables.entrySet().stream().filter((e) -%3E e.getKey().matches(action, object))%0A                .map((e) -%3E ((SubjectAndWrapper%3CT%3E) e.getValue()).subject).iterator();%0A    %7D%0A%0A    private %3CT%3E SubjectAndWrapper%3CT%3E getSubjectAndWrapper(Interest%3CT%3E interest, boolean initialize) %7B%0A        @SuppressWarnings(%22unchecked%22)%0A        SubjectAndWrapper%3CT%3E sub = (SubjectAndWrapper%3CT%3E)
  obs
@@ -1969,18 +1969,18 @@
 rvables.
-pu
+ge
 t(intere
@@ -1985,24 +1985,20 @@
 rest
-, sub
 );%0A
+%0A
         
 %7D%0A%0A 
@@ -1985,35 +1985,69 @@
 rest);%0A%0A        
-%7D%0A%0A
+if (initialize && sub == null) %7B%0A    
         Subscrip
@@ -2129,35 +2129,138 @@
 rest));%0A
-%0A        return sub
+            Subject%3CT, T%3E subject = PublishSubject.%3CT%3Ecreate().toSerialized();%0A            Observable%3CT%3E wrapper = subject
 .doOnSub
@@ -2288,16 +2288,37 @@
 cribe())
+%0A                    
 .doOnUns
@@ -2356,144 +2356,80 @@
 ));%0A
+%0A
     
-%7D%0A%0A
     
-@SuppressWarnings(%22unchecked%22)%0A
     
-public %3CT%3E Iterator%3CSubject%3CT, T%3E%3E matchingSubjects(Action%3CT%3E action, T object) %7B%0A
+sub = new SubjectAndWrapper%3C%3E(subject, wrapper);%0A  
         
 retu
@@ -2416,38 +2416,33 @@
 per);%0A          
-return
+ 
  observables.ent
@@ -2442,77 +2442,27 @@
 les.
-entrySet().stream().filter((e) -%3E e.getKey().matches(action, object))
+put(interest, sub);
 %0A   
@@ -2462,32 +2462,35 @@
 b);%0A        
+%7D%0A%0A
         
 .map((e) -%3E 
@@ -2481,60 +2481,18 @@
     
-.map((e) -%3E (Subject%3CT, T%3E) e.getValue()).iterator()
+return sub
 ;%0A  
@@ -3044,11 +3044,305 @@
  %7D%0A    %7D
+%0A%0A    private static class SubjectAndWrapper%3CT%3E %7B%0A        final Subject%3CT, T%3E subject;%0A        final Observable%3CT%3E wrapper;%0A%0A        private SubjectAndWrapper(Subject%3CT, T%3E subject, Observable%3CT%3E wrapper) %7B%0A            this.subject = subject;%0A            this.wrapper = wrapper;%0A        %7D%0A    %7D
 %0A%7D%0A

Fix observer tracking