<commit_before>using System;
using System.Linq;
using System.Reactive;
using System.Reflection;

using JetBrains.Annotations;

namespace CodeJam.Extensibility.EventBroker
{
	/// <summary>
	///      Event Broker.
	/// </summary>
	public static class EventBrokerHelper
	{
		/// <summary>
		/// Subscribe to event with delegate for OnNext.
		/// </summary>
		public static IDisposable Subscribe<T>(
			[NotNull] this IEventBroker eventBroker,
			[NotNull] string eventName,
			[NotNull] Action<T> nextAction)
		{
			if (eventBroker == null)
				throw new ArgumentNullException(nameof(eventBroker));
			if (eventName == null)
				throw new ArgumentNullException(nameof(eventName));
			if (nextAction == null)
				throw new ArgumentNullException(nameof(nextAction));

			return eventBroker.Subscribe(eventName, Observer.Create(nextAction));
		}

		/// <summary>
		///      .
		/// </summary>
		public static IDisposable SubscribeEventHandlers(
			[NotNull] object instance,
			[NotNull] IServiceProvider provider)
		{
			if (instance == null)
				throw new ArgumentNullException(nameof(instance));
			if (provider == null)
				throw new ArgumentNullException(nameof(provider));

			var eventBroker = provider.GetRequiredService<IEventBroker>();
			return 
				instance
					.GetType()
					.GetMethods(BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic)
					.SelectMany(
						method => method.GetCustomAttributes<EventHandlerAttribute>(true),
						(method, attr) =>
							WrapException(
								() => eventBroker.SubscribeMethod(attr.EventName, method, instance),
								() => $"Error subscribing method \'{method}\'."))
					.Merge();
		}

		private static T WrapException<T>(Func<T> action, Func<string> message)
		{
			try
			{
				return action();
			}
			catch (Exception ex)
			{
				throw new ApplicationException(message(), ex);
			}
		}

		/// <summary>
		///     .
		/// </summary>
		public static IDisposable RegisterEventSources(
			[NotNull] object instance,
			[NotNull] IServiceProvider provider)
		{
			if (instance == null)
				throw new ArgumentNullException(nameof(instance));
			if (provider == null)
				throw new ArgumentNullException(nameof(provider));

			var eventBroker = provider.GetRequiredService<IEventBroker>();

			var instanceType = instance.GetType();
			return
				instanceType
					.GetProperties(BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic)
					.SelectMany(
						property => property.GetCustomAttributes<EventSourceAttribute>(true),
						(property, attr) =>
							WrapException(
								() => eventBroker.RegisterProperty(attr.EventName, property, instance),
								() => $"Error while registring property \'{property}\'."))
						.Concat(
							instanceType
								.GetEvents(BindingFlags.Instance | BindingFlags.Public)
								.SelectMany(
									ev => ev.GetCustomAttributes<EventSourceAttribute>(true),
									(ev, attr) =>
										WrapException(
											() => eventBroker.RegisterEvent(attr.EventName, ev, instance),
											() => $"Error while registering event \'{ev}\'.")))
						.Merge();
		}

		private static IDisposable SubscribeMethod(
			this IEventBroker eventBroker,
			string eventName,
			MethodInfo methodInfo,
			object instance)
		{
			var parameters = methodInfo.GetParameters();
			if (parameters.Length != 1)
				throw new ApplicationException("Method must have one argument.");

			var argType = parameters[0].ParameterType;
			return
				(IDisposable)Activator
					.CreateInstance(
						typeof (FromBrokerObserver<>).MakeGenericType(argType),
						instance,
						methodInfo,
						eventName,
						eventBroker);
		}

		private static IDisposable RegisterProperty(
			this IEventBroker eventBroker,
			string eventName,
			PropertyInfo property,
			object instance)
		{
			if (!property.PropertyType.IsGenericType<commit_msg>Fix bug induced by new Merge() laziness<commit_after>using System;
using System.Linq;
using System.Reactive;
using System.Reflection;

using JetBrains.Annotations;

namespace CodeJam.Extensibility.EventBroker
{
	/// <summary>
	///      Event Broker.
	/// </summary>
	public static class EventBrokerHelper
	{
		/// <summary>
		/// Subscribe to event with delegate for OnNext.
		/// </summary>
		public static IDisposable Subscribe<T>(
			[NotNull] this IEventBroker eventBroker,
			[NotNull] string eventName,
			[NotNull] Action<T> nextAction)
		{
			if (eventBroker == null)
				throw new ArgumentNullException(nameof(eventBroker));
			if (eventName == null)
				throw new ArgumentNullException(nameof(eventName));
			if (nextAction == null)
				throw new ArgumentNullException(nameof(nextAction));

			return eventBroker.Subscribe(eventName, Observer.Create(nextAction));
		}

		/// <summary>
		///      .
		/// </summary>
		public static IDisposable SubscribeEventHandlers(
			[NotNull] object instance,
			[NotNull] IServiceProvider provider)
		{
			if (instance == null)
				throw new ArgumentNullException(nameof(instance));
			if (provider == null)
				throw new ArgumentNullException(nameof(provider));

			var eventBroker = provider.GetRequiredService<IEventBroker>();
			return 
				instance
					.GetType()
					.GetMethods(BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic)
					.SelectMany(
						method => method.GetCustomAttributes<EventHandlerAttribute>(true),
						(method, attr) =>
							WrapException(
								() => eventBroker.SubscribeMethod(attr.EventName, method, instance),
								() => $"Error subscribing method \'{method}\'."))
					.ToArray() // Merge is lazy, so it need to iterate
					.Merge();
		}

		private static T WrapException<T>(Func<T> action, Func<string> message)
		{
			try
			{
				return action();
			}
			catch (Exception ex)
			{
				throw new ApplicationException(message(), ex);
			}
		}

		/// <summary>
		///     .
		/// </summary>
		public static IDisposable RegisterEventSources(
			[NotNull] object instance,
			[NotNull] IServiceProvider provider)
		{
			if (instance == null)
				throw new ArgumentNullException(nameof(instance));
			if (provider == null)
				throw new ArgumentNullException(nameof(provider));

			var eventBroker = provider.GetRequiredService<IEventBroker>();

			var instanceType = instance.GetType();
			return
				instanceType
					.GetProperties(BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic)
					.SelectMany(
						property => property.GetCustomAttributes<EventSourceAttribute>(true),
						(property, attr) =>
							WrapException(
								() => eventBroker.RegisterProperty(attr.EventName, property, instance),
								() => $"Error while registring property \'{property}\'."))
						.Concat(
							instanceType
								.GetEvents(BindingFlags.Instance | BindingFlags.Public)
								.SelectMany(
									ev => ev.GetCustomAttributes<EventSourceAttribute>(),
									(ev, attr) =>
										WrapException(
											() => eventBroker.RegisterEvent(attr.EventName, ev, instance),
											() => $"Error while registering event \'{ev}\'.")))
						.ToArray() // Merge is lazy, so it need to iterate
						.Merge();
		}

		private static IDisposable SubscribeMethod(
			this IEventBroker eventBroker,
			string eventName,
			MethodInfo methodInfo,
			object instance)
		{
			var parameters = methodInfo.GetParameters();
			if (parameters.Length != 1)
				throw new ApplicationException("Method must have one argument.");

			var argType = parameters[0].ParameterType;
			return
				(IDisposable)Activator
					.CreateInstance(
						typeof (FromBrokerObserver<>).MakeGenericType(argType),
						instance,
						methodInfo,
						eventName,
						eventBroker);
		}

		private static IDisposable RegisterProperty(
			this IEventBroker eventBroker,
			string eventName,
			PropertyInfo property,
			object instance)
		{
			if (!property.PropertyType.IsGenericType
{'fraction_non_alphanumeric': 0.11551702154787709, 'fraction_numerical': 0.0005100089251561902, 'mean_word_length': 3.3146314631463145, 'pattern_counts': {'":': 0, '<': 41, '<?xml version=': 0, '>': 65, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '3730138', 'n_tokens_mistral': 2935, 'n_tokens_neox': 2199, 'n_words': 521}