RX-Practical

gistfile1.cs

using(var pub = new ChangeReceiver("tcp://*:5555"))
{
    Console.WriteLine("Listening...");

    var staffSender = new NotificationSender("tcp://localhost:5556");
    var customerSender = new NotificationSender("tcp://localhost:5557");

    var obs = Observable.FromEventPattern<Tuple<Guid, string>>(pub, "ChangeRecieved").Select(ep => ep.EventArgs);

    obs.GroupBy(x => x.Item2)
        .Subscribe(o =>
        {
            o.Where(x => x.Item2 == "StaffOnly!" || x.Item2 == "Change!")
                .Select(x => x.Item1)
                .Subscribe(staffSender.Send);
            o.Where(x => x.Item2 == "CustomerOnly!" || x.Item2 == "Change!")
                .Select(t => t.Item1)
                .Delay(TimeSpan.FromSeconds(6))
                .Subscribe(customerSender.Send);
        });

    pub.Start();
    Console.ReadLine();
    Console.WriteLine("Closing down.");
}