initial net work

Selector.Net/BaseNode.cs

@@ -0,0 +1,9 @@
+using System;
+
+namespace Selector.Net;
+
+public class BaseNode<T> : INode<T>
+{
+    public T Id { get; set; }
+}
+

Selector.Net/BaseSinkSource.cs

@@ -0,0 +1,19 @@
+using System;
+
+namespace Selector.Net;
+
+public abstract class BaseSinkSource<TNodeId, TObj> : BaseSingleSink<TNodeId, TObj>, ISource<TNodeId>
+{
+    protected Emit<TNodeId> EmitHandler { get; set; }
+
+    public void ReceiveHandler(Emit<TNodeId> handler)
+    {
+        EmitHandler = handler;
+    }
+
+    protected void Emit(object obj)
+    {
+        EmitHandler(this, obj);
+    }
+}
+

Selector.Net/Graph.cs

@@ -0,0 +1,135 @@
+using System;
+using QuikGraph;
+
+namespace Selector.Net
+{
+    public class Graph<TNodeId> : IGraph<TNodeId>
+    {
+        protected AdjacencyGraph<INode<TNodeId>, SEdge<INode<TNodeId>>> graph { get; set; }
+        private readonly object graphLock = new object();
+
+        public Graph()
+        {
+            graph = new();
+        }
+
+        public IEnumerable<INode<TNodeId>> Nodes => graph.Vertices;
+
+        public Task AddEdge(INode<TNodeId> from, INode<TNodeId> to)
+        {
+            lock(graphLock)
+            {
+                graph.AddVerticesAndEdge(new SEdge<INode<TNodeId>>(from, to));
+            }
+
+            if (from is ISource<TNodeId> fromSource)
+            {
+                fromSource.ReceiveHandler(SourceHandler);
+            }
+
+            if (to is ISource<TNodeId> toSource)
+            {
+                toSource.ReceiveHandler(SourceHandler);
+            }
+
+            return Task.CompletedTask;
+        }
+
+        public Task AddNode(INode<TNodeId> node)
+        {
+            lock (graphLock)
+            {
+                graph.AddVertex(node);
+            }
+
+            if (node is ISource<TNodeId> source)
+            {
+                source.ReceiveHandler(SourceHandler);
+            }
+
+            return Task.CompletedTask;
+        }
+
+        private async void SourceHandler(ISource<TNodeId> sender, object obj,
+            IEnumerable<TNodeId> nodeWhitelist = null, IEnumerable<TNodeId> nodeBlacklist = null)
+        {
+            if(nodeWhitelist is not null && nodeBlacklist is not null && nodeWhitelist.Any() && nodeBlacklist.Any())
+            {
+                throw new ArgumentException("Cannot provide whitelist and blacklist, at most one");
+            }
+
+            if (graph.TryGetOutEdges(sender, out var edges))
+            {
+                foreach (var edge in edges)
+                {
+                    if (edge.Target is ISink<TNodeId> sink)
+                    {
+                        if(nodeWhitelist is not null && nodeWhitelist.Any())
+                        {
+                            if(nodeWhitelist.Contains(sink.Id))
+                            {
+                                await sink.Consume(obj);
+                            }
+                        }
+                        else if (nodeBlacklist is not null && nodeBlacklist.Any())
+                        {
+                            if (!nodeBlacklist.Contains(sink.Id))
+                            {
+                                await sink.Consume(obj);
+                            }
+                        }
+                        else
+                        {
+                            await sink.Consume(obj);
+                        }
+                    }
+                }
+            }
+        }
+
+        public IEnumerable<ISink<TNodeId>> GetSinks()
+        {
+            foreach (var node in graph.Vertices)
+            {
+                if (node is ISink<TNodeId> sink)
+                {
+                    yield return sink;
+                }
+            }
+        }
+
+        public IEnumerable<ISink<TNodeId, TSink>> GetSinks<TSink>()
+        {
+            foreach (var node in graph.Vertices)
+            {
+                if (node is ISink<TNodeId, TSink> sink)
+                {
+                    yield return sink;
+                }
+            }
+        }
+
+        public IEnumerable<ISource<TNodeId>> GetSources()
+        {
+            foreach (var node in graph.Vertices)
+            {
+                if (node is ISource<TNodeId> source)
+                {
+                    yield return source;
+                }
+            }
+        }
+
+        public async Task Sink(string topic, object obj)
+        {
+            foreach (var sink in GetSinks())
+            {
+                if (sink.Topics.Contains(topic))
+                {
+                    await sink.Consume(obj);
+                }
+            }
+        }
+    }
+}
+

Selector.Net/Interfaces/IGraph.cs

@@ -0,0 +1,21 @@
+using System;
+using QuikGraph;
+
+namespace Selector.Net
+{
+    public interface IGraph<TNodeId>
+    {
+        IEnumerable<INode<TNodeId>> Nodes { get; }
+
+        Task AddEdge(INode<TNodeId> from, INode<TNodeId> to);
+        Task AddNode(INode<TNodeId> node);
+
+        Task Sink(string topic, object obj);
+
+        IEnumerable<ISource<TNodeId>> GetSources();
+
+        IEnumerable<ISink<TNodeId>> GetSinks();
+        IEnumerable<ISink<TNodeId, TSink>> GetSinks<TSink>();
+    }
+}
+

Selector.Net/Interfaces/INode.cs

@@ -0,0 +1,10 @@
+using System;
+
+namespace Selector.Net
+{
+    public interface INode<T>
+    {
+        public T Id { get; set; }
+    }
+}
+

Selector.Net/Interfaces/ISink.cs

@@ -0,0 +1,25 @@
+using System;
+namespace Selector.Net
+{
+    public interface ISink<TNodeId> : INode<TNodeId>
+    {
+        IEnumerable<string> Topics { get; set; }
+
+        Task Consume(object obj);
+    }
+
+    /// <summary>
+    /// Not a node, just callback handler
+    /// </summary>
+    /// <typeparam name="TObj"></typeparam>
+    public interface ITypeSink<TObj>
+    {
+        Task ConsumeType(TObj obj);
+    }
+
+    public interface ISink<TNodeId, TObj> : ISink<TNodeId>, ITypeSink<TObj>
+    {
+        //Task ConsumeType(TObj obj);
+    }
+}
+

Selector.Net/Interfaces/ISource.cs

@@ -0,0 +1,17 @@
+using System;
+namespace Selector.Net
+{
+    public delegate void Emit<T>(ISource<T> sender, object obj, IEnumerable<T> nodeWhitelist = null, IEnumerable<T> nodeBlacklist = null);
+    public delegate void Emit<TNodeId, TObj>(ISource<TNodeId, TObj> sender, object obj, IEnumerable<T> nodeWhitelist = null, IEnumerable<T> nodeBlacklist = null);
+
+    public interface ISource<TNodeId> : INode<TNodeId>
+    {
+        void ReceiveHandler(Emit<TNodeId> handler);
+    }
+
+    public interface ISource<TNodeId, TObj> : INode<TNodeId>
+    {
+        void ReceiveHandler(Emit<TNodeId, TObj> handler);
+    }
+}
+

Selector.Net/Playlist/Added.cs

@@ -0,0 +1,56 @@
+using System;
+using SpotifyAPI.Web;
+
+namespace Selector.Net.Playlist;
+
+public enum AddedType
+{
+    Since, Before
+}
+
+
+public class Added<TNodeId> : BaseSinkSource<TNodeId, PlaylistChangeEventArgs>
+{
+    public DateTime Threshold { get; set; }
+    public bool IncludeNull { get; set; }
+
+    public AddedType Operator { get; set; } = AddedType.Since;
+
+    private IEnumerable<PlaylistTrack<IPlayableItem>> Filter(IEnumerable<PlaylistTrack<IPlayableItem>> tracks)
+    {
+        return tracks.Where(t =>
+        {
+            if (t.AddedAt is not null)
+            {
+                switch (Operator)
+                {
+                    case AddedType.Since:
+                        return t.AddedAt.Value > Threshold;
+                    case AddedType.Before:
+                        return t.AddedAt.Value < Threshold;
+                }
+            }
+            else
+            {
+                if (IncludeNull)
+                {
+                    return true;
+                }
+            }
+
+            return false;
+        });
+    }
+
+    public override Task ConsumeType(PlaylistChangeEventArgs obj)
+    {
+        //obj.CurrentTracks = Filter(obj.CurrentTracks);
+        obj.AddedTracks = Filter(obj.AddedTracks);
+        obj.RemovedTracks = Filter(obj.RemovedTracks);
+
+        Emit(obj);
+
+        return Task.CompletedTask;
+    }
+}
+

Selector.Net/Playlist/Aggregator.cs

@@ -0,0 +1,60 @@
+using System;
+using System.Linq;
+using Microsoft.Extensions.Logging;
+using SpotifyAPI.Web;
+
+namespace Selector.Net.Playlist
+{
+    public class AggregatorConfig
+    {
+        public string PlaylistId { get; set; }
+    }
+
+    public class Aggregator<TNodeId>: BaseSingleSink<TNodeId, PlaylistChangeEventArgs>
+    {
+        private readonly ILogger<Aggregator<TNodeId>> Logger;
+        private readonly ISpotifyClient SpotifyClient;
+        private readonly ICurrentItemListResolver ItemResolver;
+        private readonly AggregatorConfig Config;
+
+        public Aggregator(ISpotifyClient spotifyClient, ICurrentItemListResolver itemResolver, AggregatorConfig config, ILogger<Aggregator<TNodeId>> logger)
+        {
+            Logger = logger;
+            SpotifyClient = spotifyClient;
+            ItemResolver = itemResolver;
+            Config = config;
+        }
+
+        public override async Task ConsumeType(PlaylistChangeEventArgs obj)
+        {
+            try
+            {
+                var addedTracks = obj.AddedTracks.ToList();
+                var removedTracks = obj.AddedTracks.ToList();
+                var removedTrackURIs = obj.AddedTracks.ToList();
+
+                var currentTracks = await ItemResolver.GetCurrentItems().ConfigureAwait(false);
+
+                currentTracks = currentTracks.Where(t => addedTracks);
+                currentTracks = currentTracks.Concat(addedTracks);
+
+            }
+            catch (APIUnauthorizedException e)
+            {
+                Logger.LogDebug("Unauthorised error: [{message}] (should be refreshed and retried?)", e.Message);
+                //throw e;
+            }
+            catch (APITooManyRequestsException e)
+            {
+                Logger.LogDebug("Too many requests error: [{message}]", e.Message);
+                // throw e;
+            }
+            catch (APIException e)
+            {
+                Logger.LogDebug("API error: [{message}]", e.Message);
+                // throw e;
+            }
+        }
+    }
+}
+

Selector.Net/Playlist/Applier.cs

@@ -0,0 +1,54 @@
+using System;
+using Microsoft.Extensions.Logging;
+using SpotifyAPI.Web;
+
+namespace Selector.Net.Playlist
+{
+    public class ApplierConfig {
+        public string PlaylistId { get; set; }
+    }
+
+    public class Applier<TNodeId>: BaseSingleSink<TNodeId, PlaylistChangeEventArgs>
+    {
+        private readonly ILogger<Applier<TNodeId>> Logger;
+        private readonly ISpotifyClient SpotifyClient;
+        private readonly ApplierConfig Config;
+
+        private FullPlaylist Playlist { get; set; }
+        private IEnumerable<PlaylistTrack<IPlayableItem>> Items { get; set; }
+
+        public Applier(ISpotifyClient spotifyClient, ApplierConfig config, ILogger<Applier<TNodeId>> logger)
+        {
+            Logger = logger;
+            SpotifyClient = spotifyClient;
+            Config = config;
+        }
+
+        public override async Task ConsumeType(PlaylistChangeEventArgs obj)
+        {
+            try
+            {
+                var tracks = obj.CurrentTracks.ToList();
+                var trackUris = tracks.Select(t => t.GetUri()).ToList();
+
+                await SpotifyClient.Playlists.ReplaceItems(Config.PlaylistId, new PlaylistReplaceItemsRequest(trackUris));
+            }
+            catch (APIUnauthorizedException e)
+            {
+                Logger.LogDebug("Unauthorised error: [{message}] (should be refreshed and retried?)", e.Message);
+                //throw e;
+            }
+            catch (APITooManyRequestsException e)
+            {
+                Logger.LogDebug("Too many requests error: [{message}]", e.Message);
+                // throw e;
+            }
+            catch (APIException e)
+            {
+                Logger.LogDebug("API error: [{message}]", e.Message);
+                // throw e;
+            }
+        }
+    }
+}
+

Selector.Net/Playlist/CurrentTrackResolver.cs

@@ -0,0 +1,11 @@
+using System;
+using SpotifyAPI.Web;
+
+namespace Selector.Net.Playlist
+{
+    public interface ICurrentItemListResolver
+    {
+        Task<IEnumerable<PlaylistTrack<IPlayableItem>>> GetCurrentItems();
+    }
+}
+

Selector.Net/Playlist/ItemFilter.cs

@@ -0,0 +1,26 @@
+using System;
+using SpotifyAPI.Web;
+
+namespace Selector.Net.Playlist;
+
+public class ItemFilter<TNodeId> : BaseSinkSource<TNodeId, PlaylistChangeEventArgs>
+{
+    public Func<PlaylistTrack<IPlayableItem>, bool> Func { get; set; }
+
+    public ItemFilter(Func<PlaylistTrack<IPlayableItem>, bool> func)
+    {
+        Func = func;
+    }
+
+    public override Task ConsumeType(PlaylistChangeEventArgs obj)
+    {
+        //obj.CurrentTracks = obj.CurrentTracks.Where(Func);
+        obj.AddedTracks = obj.AddedTracks.Where(Func);
+        obj.RemovedTracks = obj.RemovedTracks.Where(Func);
+
+        Emit(obj);
+
+        return Task.CompletedTask;
+    }
+}
+

Selector.Net/Playlist/ItemMutator.cs

@@ -0,0 +1,26 @@
+using System;
+using SpotifyAPI.Web;
+
+namespace Selector.Net.Playlist;
+
+public class ItemMutator<TNodeId> : BaseSinkSource<TNodeId, PlaylistChangeEventArgs>
+{
+    public Func<PlaylistTrack<IPlayableItem>, PlaylistTrack<IPlayableItem>> Func { get; set; }
+
+    public ItemMutator(Func<PlaylistTrack<IPlayableItem>, PlaylistTrack<IPlayableItem>> func)
+    {
+        Func = func;
+    }
+
+    public override Task ConsumeType(PlaylistChangeEventArgs obj)
+    {
+        //obj.CurrentTracks = obj.CurrentTracks.Select(Func);
+        obj.AddedTracks = obj.AddedTracks.Select(Func);
+        obj.RemovedTracks = obj.RemovedTracks.Select(Func);
+
+        Emit(obj);
+
+        return Task.CompletedTask;
+    }
+}
+

Selector.Net/Playlist/PlaylistFilter.cs

@@ -0,0 +1,68 @@
+using System;
+
+namespace Selector.Net.Playlist
+{
+    public class PlaylistFilterConfig
+    {
+        public IEnumerable<string> NameWhiteList { get; set; }
+        public IEnumerable<string> NameBlackList { get; set; }
+        public IEnumerable<string> UriWhiteList { get; set; }
+        public IEnumerable<string> UriBlackList { get; set; }
+    }
+
+    public class PlaylistFilter<TNodeId> : BaseSinkSource<TNodeId, PlaylistChangeEventArgs>
+    {
+        public IEnumerable<string> NameWhiteList { get; set; }
+        public IEnumerable<string> NameBlackList { get; set; }
+        public IEnumerable<string> UriWhiteList { get; set; }
+        public IEnumerable<string> UriBlackList { get; set; }
+
+        public PlaylistFilter() { }
+
+        public PlaylistFilter(PlaylistFilterConfig config)
+        {
+            NameWhiteList = config.NameWhiteList;
+            NameBlackList = config.NameBlackList;
+            UriWhiteList = config.UriWhiteList;
+            UriBlackList = config.UriBlackList;
+        }
+
+        public override Task ConsumeType(PlaylistChangeEventArgs obj)
+        {
+            if (NameWhiteList is not null && NameWhiteList.Any())
+            {
+                if (NameWhiteList.Contains(obj.Current.Name))
+                {
+                    Emit(obj);
+                }
+            }
+
+            if (NameBlackList is not null && NameBlackList.Any())
+            {
+                if (!NameBlackList.Contains(obj.Current.Name))
+                {
+                    Emit(obj);
+                }
+            }
+
+            if (UriWhiteList is not null && UriWhiteList.Any())
+            {
+                if (UriWhiteList.Contains(obj.Current.Name))
+                {
+                    Emit(obj);
+                }
+            }
+
+            if (UriBlackList is not null && UriBlackList.Any())
+            {
+                if (!UriBlackList.Contains(obj.Current.Name))
+                {
+                    Emit(obj);
+                }
+            }
+
+            return Task.CompletedTask;
+        }
+    }
+}
+

Selector.Net/Playlist/TypeFilter.cs

@@ -0,0 +1,40 @@
+using System;
+using SpotifyAPI.Web;
+
+namespace Selector.Net.Playlist;
+
+public enum PlayableItemType
+{
+    Track, Episode
+}
+
+public class TypeFilter<TNodeId> : BaseSinkSource<TNodeId, IEnumerable<PlaylistTrack<IPlayableItem>>>
+{
+    public PlayableItemType FilterType { get; set; }
+
+    public TypeFilter(PlayableItemType filterType)
+    {
+        FilterType = filterType;
+    }
+
+    public override Task ConsumeType(IEnumerable<PlaylistTrack<IPlayableItem>> tracks)
+    {
+        switch (FilterType)
+        {
+            case PlayableItemType.Track:
+
+                Emit(tracks.Where(i => i.Track is FullTrack));
+
+                break;
+            case PlayableItemType.Episode:
+
+                Emit(tracks.Where(i => i.Track is FullEpisode));
+
+                break;
+        }
+
+
+        return Task.CompletedTask;
+    }
+}
+

Selector.Net/PlaylistGraph.cs

@@ -0,0 +1,18 @@
+using System;
+namespace Selector.Net.Playlist;
+
+public class PlaylistGraph<TNodeId>
+{
+    private IGraph<TNodeId> graph { get; set; }
+
+    public PlaylistGraph(PlaylistFilterConfig filterConfig)
+    {
+        graph = new Graph<TNodeId>();
+
+        var entryFilter = new PlaylistFilter<TNodeId>(filterConfig)
+        {
+            Topics = new[] { "track-entry" }
+        };
+    }
+}
+

Selector.Net/Repeater.cs

@@ -0,0 +1,18 @@
+using System;
+
+namespace Selector.Net;
+
+public class Repeater<TNodeId>: BaseSource<TNodeId>, ISink<TNodeId>
+{
+    public IEnumerable<string> Topics { get; set; }
+
+    private Type[] _types = Array.Empty<Type>();
+    public IEnumerable<Type> Types => _types;
+
+    public Task Consume(object obj)
+    {
+        Emit(obj);
+
+        return Task.CompletedTask;
+    }
+}

Selector.Net/Selector.Net.csproj

@@ -0,0 +1,32 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <TargetFramework>net6.0</TargetFramework>
+    <ImplicitUsings>enable</ImplicitUsings>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <None Remove="QuikGraph" />
+    <None Remove="Interfaces\" />
+    <None Remove="Source\" />
+    <None Remove="Sink\" />
+    <None Remove="Playlist\" />
+    <None Remove="Microsoft.Extensions.Logging.Abstractions" />
+    <None Remove="System.Linq.Async" />
+  </ItemGroup>
+  <ItemGroup>
+    <PackageReference Include="QuikGraph" Version="2.3.0" />
+    <PackageReference Include="SpotifyAPI.Web" Version="6.2.2" />
+    <PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="6.0.1" />
+    <PackageReference Include="System.Linq.Async" Version="6.0.1" />
+  </ItemGroup>
+  <ItemGroup>
+    <Folder Include="Interfaces\" />
+    <Folder Include="Source\" />
+    <Folder Include="Sink\" />
+    <Folder Include="Playlist\" />
+  </ItemGroup>
+  <ItemGroup>
+    <ProjectReference Include="..\Selector\Selector.csproj" />
+  </ItemGroup>
+</Project>

Selector.Net/Sink/BaseMultiSink.cs

@@ -0,0 +1,30 @@
+using System;
+
+namespace Selector.Net;
+
+//public abstract class BaseMultiSink<TNodeId, TObj>: BaseNode<TNodeId>, ISink<TNodeId>
+//{
+//    public IEnumerable<string> Topics { get; }
+
+//    public IDictionary<Type, Action<object>> Callbacks { get; set; }
+
+
+//    public Task Consume(object obj)
+//    {
+//        var objType = obj.GetType();
+
+//        if(Callbacks.ContainsKey(objType))
+//        {
+//            var callback = Callbacks[objType];
+
+//            var objCast = (TObj)obj;
+
+//            return ConsumeType(objCast);
+//        }
+//        else
+//        {
+//            throw new ArgumentException("Not of acceptable payload type");
+//        }
+//    }
+//}
+

Selector.Net/Sink/BaseSingleSink.cs

@@ -0,0 +1,27 @@
+using System;
+
+namespace Selector.Net;
+
+public abstract class BaseSingleSink<TNodeId, TObj>: BaseNode<TNodeId>, ISink<TNodeId, TObj>
+{
+    public IEnumerable<string> Topics { get; set; }
+
+    public Task Consume(object obj)
+    {
+        var type = GetType();
+        var genericArgs = type.GetGenericArguments();
+        var objType = obj.GetType();
+
+        if (objType.IsAssignableTo(genericArgs[1]))
+        {
+            var objCast = (TObj)obj;
+
+            return ConsumeType(objCast);
+        }
+
+        return Task.CompletedTask;
+    }
+
+    public abstract Task ConsumeType(TObj obj);
+}
+

Selector.Net/Sink/DebugSink.cs

@@ -0,0 +1,28 @@
+using System;
+using Microsoft.Extensions.Logging;
+
+namespace Selector.Net;
+
+public class DebugSink<TNodeId>: ISink<TNodeId>
+{
+    public IEnumerable<string> Topics { get; set; }
+    public TNodeId Id { get; set; }
+
+    public ILogger<DebugSink<TNodeId>> Logger { get; set; }
+
+    public Task Consume(object obj)
+    {
+        var type = obj.GetType();
+
+        if (Logger is not null)
+        {
+            Logger.LogDebug("{} Received, {}", type, obj);
+        }else
+        {
+            Console.WriteLine("{0} Received, {1}", type, obj);
+        }
+
+        return Task.CompletedTask;
+    }
+}
+

Selector.Net/Sink/EmptySink.cs

@@ -0,0 +1,12 @@
+using System;
+
+namespace Selector.Net;
+
+public class EmptySink<TNodeId, TObj>: BaseSingleSink<TNodeId, TObj>
+{
+    public override Task ConsumeType(TObj obj)
+    {
+        return Task.CompletedTask;
+    }
+}
+

Selector.Net/Source/BaseSource.cs

@@ -0,0 +1,19 @@
+using System;
+
+namespace Selector.Net;
+
+public abstract class BaseSource<TNodeId>: BaseNode<TNodeId>, ISource<TNodeId>
+{
+    protected Emit<TNodeId> EmitHandler { get; set; }
+
+    public void ReceiveHandler(Emit<TNodeId> handler)
+    {
+        EmitHandler = handler;
+    }
+
+    protected void Emit(object obj)
+    {
+        EmitHandler(this, obj);
+    }
+}
+

Selector.Net/Source/TriggerSource.cs

@@ -0,0 +1,10 @@
+namespace Selector.Net;
+
+public class TriggerSource<T> : BaseSource<T>
+{
+    public void Trigger(T obj)
+    {
+        Emit(obj);
+    }
+}
+

Selector.Tests/Net/Graph.cs

@@ -0,0 +1,127 @@
+using System;
+using System.Collections.Generic;
+using QuikGraph;
+using Selector.Net;
+using Xunit;
+using FluentAssertions;
+using Moq;
+
+namespace Selector.Tests.Net
+{
+    public class GraphTests
+    {
+        [Fact]
+        public void Network()
+        {
+            var net = new Graph<string>();
+
+            var trigger = new TriggerSource<string>()
+            {
+                Id = "trigger"
+            };
+
+            var sink = new EmptySink<string, object>()
+            {
+                Id = "sink"
+            };
+
+            net.AddEdge(
+                trigger,
+                sink
+            );
+
+            trigger.Trigger("payload");
+        }
+
+        [Fact]
+        public void SourceReceivesPayload()
+        {
+            var net = new Graph<string>();
+
+            var trigger = new TriggerSource<string>()
+            {
+                Id = "trigger"
+            };
+
+            var sink = new Mock<ISink<string>>();
+
+            net.AddEdge(
+                trigger,
+                sink.Object
+            );
+
+            var payload = "payload";
+
+            trigger.Trigger(payload);
+
+            sink.Verify(a => a.Consume(payload), Times.Once());
+            sink.VerifyNoOtherCalls();
+        }
+
+        [Fact]
+        public void SourceReceivesGraphPayload()
+        {
+            var net = new Graph<string>();
+
+            var sink = new Mock<ISink<string>>();
+            sink.Setup(a => a.Topics).Returns(new[] {"topic1"});
+
+            net.AddNode(
+                sink.Object
+            );
+
+            var payload = "payload";
+
+            net.Sink("topic1", payload);
+
+            sink.Verify(a => a.Consume(payload), Times.Once());
+            sink.Verify(a => a.Topics, Times.Once());
+            sink.VerifyNoOtherCalls();
+        }
+
+        [Fact]
+        public void SourceReceivesRepeatedPayload()
+        {
+            var net = new Graph<string>();
+
+            var trigger = new TriggerSource<string>()
+            {
+                Id = "trigger"
+            };
+
+            var repeater = new Repeater<string>();
+            var repeater2 = new Repeater<string>();
+            var repeater3 = new Repeater<string>();
+
+            var sink = new Mock<ISink<string>>();
+
+            net.AddEdge(
+                trigger,
+                repeater
+            );
+
+            net.AddEdge(
+                repeater,
+                repeater2
+            );
+
+            net.AddEdge(
+                repeater2,
+                repeater3
+            );
+
+            net.AddEdge(
+                repeater3,
+                sink.Object
+            );
+
+            var payload = "payload";
+
+            trigger.Trigger(payload);
+
+            sink.Verify(a => a.Consume(payload), Times.Once());
+            sink.VerifyNoOtherCalls();
+        }
+    }
+}
+

Selector.Tests/Selector.Tests.csproj

@@ -24,6 +24,13 @@
 
   <ItemGroup>
     <ProjectReference Include="..\Selector\Selector.csproj" />
+    <ProjectReference Include="..\Selector.Net\Selector.Net.csproj" />
   </ItemGroup>
 
+  <ItemGroup>
+    <None Remove="Net\" />
+  </ItemGroup>
+  <ItemGroup>
+    <Folder Include="Net\" />
+  </ItemGroup>
 </Project>

Selector.Tests/Watcher/PlayerWatcher.cs

@@ -9,8 +9,8 @@ using System.Threading;
 using System.Threading.Tasks;
 using Xunit.Sdk;
 
-namespace Selector.Tests
-{
+namespace Selector.Tests;
+
 public class PlayerWatcherTests
 {
     public static IEnumerable<object[]> NowPlayingData =>
@@ -238,4 +238,3 @@ namespace Selector.Tests
     //     await watch.Watch(token.Token);
     // }
 }
-}

Selector.sln

@@ -24,6 +24,8 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Selector.Cache", "Selector.
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Selector.Event", "Selector.Event\Selector.Event.csproj", "{C2FF1673-CB1A-43B7-A814-07BB3CB3A0D6}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Selector.Net", "Selector.Net\Selector.Net.csproj", "{825F16A4-DB67-4CC4-A4D0-8326D297E227}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -58,6 +60,10 @@ Global
 		{C2FF1673-CB1A-43B7-A814-07BB3CB3A0D6}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{C2FF1673-CB1A-43B7-A814-07BB3CB3A0D6}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{C2FF1673-CB1A-43B7-A814-07BB3CB3A0D6}.Release|Any CPU.Build.0 = Release|Any CPU
+		{825F16A4-DB67-4CC4-A4D0-8326D297E227}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{825F16A4-DB67-4CC4-A4D0-8326D297E227}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{825F16A4-DB67-4CC4-A4D0-8326D297E227}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{825F16A4-DB67-4CC4-A4D0-8326D297E227}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE