Livehub

livehub.cs

void Main() {
    Util.CurrentQueryPath.Dump();
    string url = "http://localhost:8080";


    using (WebApp.Start(url, Configuration)) {
        //Task.Run(DoTask);

        Console.ReadLine();
    }
}

public void Configuration(IAppBuilder app) {
    var rootPath = @"C:\Projects\LINQPad\Journal\2017-10\www\";

    app.UseCors(CorsOptions.AllowAll);
    app.MapSignalR();
    app.UseFileServer(new FileServerOptions {

        EnableDefaultFiles = true,
        EnableDirectoryBrowsing = true,
        FileSystem = new PhysicalFileSystem(Path.Combine(rootPath, "Scripts")),
        RequestPath = new PathString("/Scripts")
    });
    app.Use(async (IOwinContext ctx, Func<Task> next) => {

        var html = File.ReadAllText(Path.Combine(rootPath, "index.html"));

        await ctx.Response.WriteAsync(html);

        await next();

    });

}

public class LiveHub {

    Func<LiveResource, Task> OnGotoPage;

    public LiveHub(Func<LiveResource, Task> onGotoPage) {
        OnGotoPage = onGotoPage;
    }

    //  Connection -> Resource
    private ConcurrentDictionary<string, string> Subscriptions = new ConcurrentDictionary<string, string>();

    //  Track all the live resources by path
    private ConcurrentDictionary<string, LiveResource> LiveResources = new ConcurrentDictionary<string, LiveResource>();


    public void Goto(string connectionId, string path) {
        //  Set the current subscription to the new path
        Subscriptions.AddOrUpdate(connectionId, (c) => path, (c, op) => {
            LiveResource oldResource;
            if (LiveResources.TryGetValue(path, out oldResource)) {
                oldResource.RemoveConnection(connectionId);
            }
            return path;
        });
        

        var resource =
            LiveResources.AddOrUpdate(
                key: path,
                addValueFactory: (p) => {
                    var liveResource = new LiveResource(connectionId, path, OnGotoPage);
                    liveResource.Run();
                    return liveResource;
                },
                updateValueFactory: (p, r) => r
            );

        resource.AddConnection(connectionId);


        

    }

    public void Disconnect(string connectionId) {
        string path;
        if (Subscriptions.TryGetValue(connectionId, out path)) {
            LiveResource resource;
            if (LiveResources.TryGetValue(path, out resource)) {
                resource.RemoveConnection(connectionId);
            }
        }
        Console.WriteLine("Disconnecting {0}", connectionId);
    }


    public void Post(string connectionId, string path, string payload) {
        Console.WriteLine("Post to {0} from {1}", path, connectionId);
        Console.WriteLine("Payload: {0}", payload);
    }
}

public class Client {
    public string ConnectionId { get; private set; }
    private dynamic ClientConnection = null;

    public Client(string connectionId) {
        ConnectionId = connectionId;
    }

    public void Send(string html) {
        if (ClientConnection == null) {
            var context = GlobalHost.ConnectionManager.GetHubContext<MyHub>();
            ClientConnection = context.Clients.Client(ConnectionId);
        }
        ClientConnection.appendOutput(html);
    }
}

public class LiveResource {
    public string ConnectionId { get; private set; }
    public string Path { get; private set; }
    Func<LiveResource, Task> GetTask;
    private Task Task;
    public bool IsLive => Task?.Status == TaskStatus.Running;

    private ConcurrentDictionary<string, Client> ClientLookup = new ConcurrentDictionary<string, Client>();

    public LiveResource(string connectionId, string path, Func<LiveResource, Task> getTask) {
        ConnectionId = connectionId;
        Path = path;
        GetTask = getTask;
    }

    public void AddConnection(string id) => ClientLookup.GetOrAdd(id, (cid) => new Client(id));
    public void RemoveConnection(string id) {
        Client client;
        ClientLookup.TryRemove(id, out client);
    }
    
    public Client[] Clients => ClientLookup.Values.ToArray();

    public async Task Run() {
        if (Task == null || IsLive == false) {
            Task = GetTask(this);
        }
        await Task.Delay(10);
        try {
            await Task;
        } catch (Exception ex) {

            throw;
        }
        Close();
    }


    public void Close() {
        Console.WriteLine("Closing Resource: {0}", this.Path);
    }
}



public class MyHub : Hub {
    public override Task OnConnected() {
        return base.OnConnected();
    }

    public override Task OnDisconnected(bool stopCalled) {
        GlobalHub.Disconnect(Context.ConnectionId);
        return base.OnDisconnected(stopCalled);
    }

    public void Subscribe(string url) {
        GlobalHub.Goto(Context.ConnectionId, url);
    }
}

public static LiveHub GlobalHub = new LiveHub(async (resource) => {

    var container = new DumpContainer();
    container.Dump(resource.Path);

    for (int i = 0; i < 10000; i++) {
        await Task.Delay(1000);
        var clients = resource.Clients;

        var msg = new { i, resource.ConnectionId, resource.Path, clients };
        container.Content = msg;

        foreach (var client in clients) {
            client.Send(Util.ToHtmlString(new { client.ConnectionId, resource.Path, i }));
        }
    }

});