<commit_before>/*
 * Copyright (c) 2010 <NAME>
 * 
 *    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;
	}

}<commit_msg>Handle messages over to subscribers of superclasses<commit_after>/*
 * Copyright (c) 2010 <NAME>
 * 
 *    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 = subscribersOf(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<?>> subscribersOf(Class<T> messageType) {
		List<Class<?>> types = typesCompatibleWith(messageType);
		if(types == null) {
			return null;
		}
		List<Subscriber<?>> result = new ArrayList<Subscriber<?>>();
		for(Class<?> type : types) {
			List<Subscriber<?>> list = subscribersByType.get(type);
			if(list != null) {
				synchronized (list) {
					result.addAll(list);
				}
			}
		}
		return result;
	}
	
	private List<Class<?>> typesCompatibleWith(Class<?> type) {
		List<Class<?>> list = null;
		synchronized (subscribersByType) {
			for(Class<?> candidate : subscribersByType.keySet()) {
				if(candidate.isAssignableFrom(type)) {
					if(list == null) {
						list = new ArrayList<Class<?>>();
					}
					list.add(candidate);
				}
			}
		}
		return list;
	}

}
{'fraction_non_alphanumeric': 0.13165372229588937, 'fraction_numerical': 0.0030308770600492516, 'mean_word_length': 3.981132075471698, 'pattern_counts': {'":': 0, '<': 80, '<?xml version=': 0, '>': 80, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4540952', 'n_tokens_mistral': 1851, 'n_tokens_neox': 1648, 'n_words': 439}