using System; using System.Threading; using Cysharp.Threading.Tasks; namespace MessagePipe { public static class DistributedSubscriberExtensions { // sync handler public static UniTask SubscribeAsync(this IDistributedSubscriber subscriber, TKey key, Action handler, CancellationToken cancellationToken = default) { return SubscribeAsync(subscriber, key, handler, Array.Empty>(), cancellationToken); } public static UniTask SubscribeAsync(this IDistributedSubscriber subscriber, TKey key, Action handler, MessageHandlerFilter[] filters, CancellationToken cancellationToken = default) { return subscriber.SubscribeAsync(key, new AnonymousMessageHandler(handler), filters, cancellationToken); } public static UniTask SubscribeAsync(this IDistributedSubscriber subscriber, TKey key, Action handler, Func predicate, CancellationToken cancellationToken = default) { return SubscribeAsync(subscriber, key, handler, predicate, Array.Empty>(), cancellationToken); } public static UniTask SubscribeAsync(this IDistributedSubscriber subscriber, TKey key, Action handler, Func predicate, MessageHandlerFilter[] filters, CancellationToken cancellationToken = default) { var predicateFilter = new PredicateFilter(predicate); filters = (filters.Length == 0) ? new[] { predicateFilter } : Append(filters, predicateFilter); return subscriber.SubscribeAsync(key, new AnonymousMessageHandler(handler), filters, cancellationToken); } // async handler public static UniTask SubscribeAsync(this IDistributedSubscriber subscriber, TKey key, Func handler, CancellationToken cancellationToken = default) { return SubscribeAsync(subscriber, key, handler, Array.Empty>(), cancellationToken); } public static UniTask SubscribeAsync(this IDistributedSubscriber subscriber, TKey key, Func handler, AsyncMessageHandlerFilter[] filters, CancellationToken cancellationToken = default) { return subscriber.SubscribeAsync(key, new AnonymousAsyncMessageHandler(handler), filters, cancellationToken); } public static UniTask SubscribeAsync(this IDistributedSubscriber subscriber, TKey key, Func handler, Func predicate, CancellationToken cancellationToken = default) { return SubscribeAsync(subscriber, key, handler, predicate, Array.Empty>(), cancellationToken); } public static UniTask SubscribeAsync(this IDistributedSubscriber subscriber, TKey key, Func handler, Func predicate, AsyncMessageHandlerFilter[] filters, CancellationToken cancellationToken = default) { var predicateFilter = new AsyncPredicateFilter(predicate); filters = (filters.Length == 0) ? new[] { predicateFilter } : Append(filters, predicateFilter); return subscriber.SubscribeAsync(key, new AnonymousAsyncMessageHandler(handler), filters, cancellationToken); } static T[] Append(T[] source, T item) { var dest = new T[source.Length + 1]; Array.Copy(source, 0, dest, 0, source.Length); dest[dest.Length - 1] = item; return dest; } } }