SubscriberExtension

gistfile1.cs

 public static partial class SubscriberExtension {
        private static readonly MethodInfo MessageReceivedMethod = typeof(Subscriber).GetMethod("OnMessageReceived");


        private static readonly ConcurrentDictionary<string, object> _cacheListActions =
            new ConcurrentDictionary<string, object>();
        private static readonly Type _ilistType = typeof(IList<>);
        private static readonly MethodInfo _toObjects =
            typeof(SubscriberExtension).GetMethod("ToObjects", BindingFlags.Static | BindingFlags.NonPublic);

        private static List<object> ToObjects<T>(IList<T> list) {
            return list.Cast<object>().ToList();
        }

        private static object GetListAction(Type type, Action<IList<object>> listAction) {
            object action = null;
            var key = type.FullName;
            if (listAction == null || _cacheListActions.TryGetValue(key, out action)) return action;
            var listType = _ilistType.MakeGenericType(type);
            var toObjects = _toObjects.MakeGenericMethod(type);
            var p = Expression.Parameter(listType, "msg");
            action =
                Expression.Lambda(Expression.GetActionType(listType),
                Expression.Call(
                    listAction.Target == null ? null : Expression.Constant(listAction.Target),
                    listAction.Method, Expression.Call(null, toObjects, p)), p).Compile();
            return _cacheListActions[key] = action;
        }

        public static object OnMessageReceived(this Subscriber subscriber, string transportName, Dictionary<string, object> structure,
            Action<object> messageAction = null, Action<HandleErrorArgs> errorAction = null, TimeSpan interval = default(TimeSpan), int batchSize = 1,
            bool handleInListenerThread = true, Action<IList<object>> messageActionList = null) {
            var className = String.Concat("Class", String.Join(string.Empty, structure.Select(kv => kv.Key)));
            var messageType = ReflectionHelper.GenerateClassType(className, structure);
            var genericMethod = MessageReceivedMethod.MakeGenericMethod(messageType);
            errorAction = errorAction ?? new Action<HandleErrorArgs>(errorHandler => { errorHandler.Continue = true; });
            interval = interval == default(TimeSpan) ? TimeSpan.FromMilliseconds(100) : interval;
            var handler = genericMethod.Invoke(subscriber, new object[] { transportName, messageAction, errorAction, interval, batchSize, 
                handleInListenerThread, GetListAction(messageType, messageActionList)});
            return handler;
        }
    }