gistfile1.cs

using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Net;
using System.Reactive.Linq;
using System.Text;
using System.Web;
using Newtonsoft.Json;

namespace PubNubRx
{
    public class PubNubObservable<T> : IObservable<T>
    {
        private readonly IObservable<T> observable;

        public PubNubObservable(string subscriberKey, string channel)
        {
            observable = new PubNubObservable(subscriberKey, channel)
                                        .Select(Deserialize);
        }

        public T Deserialize(string response)
        {
            var serializer = new JsonSerializer();
            byte[] byteArray = Encoding.ASCII.GetBytes( response );
            using(var stream = new MemoryStream( byteArray ))
            using(var tr = new StreamReader(stream))
            {
                return serializer.Deserialize<T>(new JsonTextReader(tr));
            }
        }

        public IDisposable Subscribe(IObserver<T> observer)
        {
            return observable.Subscribe(observer);
        }
    }
    public class PubNubObservable : IObservable<string>
    {
        private const string Origin = "http://pubsub.pubnub.com/";
        private string currentTimeToken = "0";
        
        private readonly ISubject<string> tokens = new Subject<string>();

        public PubNubObservable(string subscriberKey, string channel)
        {


            var observable = //Observable.Create<string>(obs => Request(subscriberKey, channel, obs)).Repeat();
                tokens
                .DistinctUntilChanged()
                .SelectMany(currentTimeToken =>
                                      {
                                          var urlComponents = new[]
                                                                  {
                                                                      "subscribe", subscriberKey, channel, "0",
                                                                      currentTimeToken
                                                                  }
                                              .Select(HttpUtility.UrlEncode);
                                          var url = Origin + string.Join("/", urlComponents);
                                          var request = (HttpWebRequest) WebRequest.Create(url);
                                          request.Timeout = 200000;
                                          request.ReadWriteTimeout = 200000;
                                          return Observable.FromAsyncPattern<WebResponse>(request.BeginGetResponse,
                                                                                          request.EndGetResponse)()
                                              .Select(HandleResponse);
                                      },
                                  (x, y) => new {Token = x, Response = y});

            Observable.CreateWithDisposable<string>(obs =>
                                                        {
                                                            return observable
                                                                .Subscribe(x =>
                                                                               {
                                                                                   obs.OnNext(x.Response);
                                                                                   tokens.OnNext(x.Token);
                                                                               })
                                                                ;
                                                        });
        }

        private string HandleResponse(WebResponse response)
        {
            List<object> deserialized;
            var serializer = new JsonSerializer();
            var responseStream = response.GetResponseStream();
            if (responseStream == null || !responseStream.CanRead) return null;
            using (var st = new StreamReader(responseStream))
            {
                deserialized = serializer.Deserialize<List<object>>(new JsonTextReader(st));
            }
            if (deserialized[1].ToString().Length > 0)
                currentTimeToken = deserialized[1].ToString();
            return deserialized[0].ToString();
        }

        public IDisposable Subscribe(IObserver<string> observer)
        {
            return observable.Subscribe(observer);
        }
    }
}