/*
 * Copyright (c) 2010 Célio Cidral Junior
 * 
 *    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.tomighty.bus;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.tomighty.ioc.Inject;
import org.tomighty.log.Log;

public class Bus {
	
	@Inject private Log log;
	private Map<Class<?>, List<Subscriber<?>>> subscribersByType;
	
	public Bus() {
		subscribersByType = Collections.synchronizedMap(new HashMap<Class<?>, List<Subscriber<?>>>());
	}

	public <T> void subscribe(Subscriber<T> subscriber, Class<T> messageType) {
		List<Subscriber<?>> list = subscribersByType.get(messageType);
		if(list == null) {
			list = new ArrayList<Subscriber<?>>();
			subscribersByType.put(messageType, list);
		}
		synchronized (list) {
			list.add(subscriber);
		}
	}
	
	public <T> void unsubscribe(Subscriber<T> subscriber, Class<T> messageType) {
		List<Subscriber<?>> list = subscribersByType.get(messageType);
		if(list != null) {
			synchronized (list) {
				list.remove(subscriber);
			}
		}
	}
	
	@SuppressWarnings({ "unchecked", "rawtypes" })
	public void publish(Object message) {
		Class<? extends Object> messageType = message.getClass();
		List<Subscriber<?>> list = safeListOfSubscribers(messageType);
		if(list == null) {
			return;
		}
		for(Subscriber subscriber : list) {
			try {
				subscriber.receive(message);
			} catch(Throwable error) {
				log.error("Error delivering message to subscriber: "+subscriber, error);
			}
		}
	}

	private <T> List<Subscriber<?>> safeListOfSubscribers(Class<T> messageType) {
		List<Subscriber<?>> list = subscribersByType.get(messageType);
		if(list != null) {
			synchronized (list) {
				return new ArrayList<Subscriber<?>>(list);
			}
		}
		return null;
	}

}

@@ -1860,26 +1860,16 @@
 list = s
-afeListOfS
 ubscribe
@@ -1870,16 +1870,18 @@
 scribers
+Of
 (message
@@ -2174,18 +2174,8 @@
 %3E%3E s
-afeListOfS
 ubsc
@@ -2180,16 +2180,18 @@
 scribers
+Of
 (Class%3CT
@@ -2200,32 +2200,234 @@
 messageType) %7B%0D%0A
+%09%09List%3CClass%3C?%3E%3E types = typesCompatibleWith(messageType);%0D%0A%09%09if(types == null) %7B%0D%0A%09%09%09return null;%0D%0A%09%09%7D%0D%0A%09%09List%3CSubscriber%3C?%3E%3E result = new ArrayList%3CSubscriber%3C?%3E%3E();%0D%0A%09%09for(Class%3C?%3E type : types) %7B%0D%0A%09
 %09%09List%3CSubscribe
@@ -2457,39 +2457,33 @@
 sByType.get(
-messageT
+t
 ype);%0D%0A
+%09
 %09%09if(list !=
@@ -2487,32 +2487,33 @@
  != null) %7B%0D%0A%09%09%09
+%09
 synchronized (li
@@ -2527,50 +2527,414 @@
 %09%09%09%09
-return new ArrayList%3CSubscriber%3C?%3E%3E(list);
+%09result.addAll(list);%0D%0A%09%09%09%09%7D%0D%0A%09%09%09%7D%0D%0A%09%09%7D%0D%0A%09%09return result;%0D%0A%09%7D%0D%0A%09%0D%0A%09private List%3CClass%3C?%3E%3E typesCompatibleWith(Class%3C?%3E type) %7B%0D%0A%09%09List%3CClass%3C?%3E%3E list = null;%0D%0A%09%09synchronized (subscribersByType) %7B%0D%0A%09%09%09for(Class%3C?%3E candidate : subscribersByType.keySet()) %7B%0D%0A%09%09%09%09if(candidate.isAssignableFrom(type)) %7B%0D%0A%09%09%09%09%09if(list == null) %7B%0D%0A%09%09%09%09%09%09list = new ArrayList%3CClass%3C?%3E%3E();%0D%0A%09%09%09%09%09%7D%0D%0A%09%09%09%09%09list.add(candidate);%0D%0A%09%09%09%09%7D
 %0D%0A%09%09
@@ -2947,28 +2947,28 @@
 %7D%0D%0A%09%09return 
-null
+list
 ;%0D%0A%09%7D%0D%0A%0D%0A%7D%0D%0A

Handle messages over to subscribers of superclasses