private static IObservable<byte[]> CreateDownloadObservable(Uri uri)
 {
     return Observable.Create<byte[]>(o => {
         var result = new ReplaySubject<byte[]>();
         var inner = Observable.Using(() => new WebClient(), wc => {
             var obs = Observable
                 .FromEventPattern<
                     DownloadDataCompletedEventHandler,
                     DownloadDataCompletedEventArgs>(
                         h => wc.DownloadDataCompleted += h,
                         h => wc.DownloadDataCompleted -= h)
                 .Take(1);
             wc.DownloadDataAsync(uri);
             return obs;
         }).Subscribe(ep => {
             if (ep.EventArgs.Cancelled) {
                 result.OnCompleted();
             } else {
                 if (ep.EventArgs.Error != null) {
                     result.OnError(ep.EventArgs.Error);
                 } else {
                     result.OnNext(ep.EventArgs.Result);
                     result.OnCompleted();
                 }
             }
         }, ex => {
             result.OnError(ex);
         });
         return new CompositeDisposable(inner, result.Subscribe(o));
     }).Retry(5);
 }

 public SynchronizationController(
     IScheduler scheduler, 
     IStartSynchronizing startSynchronizing,
     ITranscodingNotifications transcodingNotifications)
 {
     if (transcodingNotifications == null) throw new ArgumentNullException(nameof(transcodingNotifications), $"{nameof(transcodingNotifications)} is null.");
     if (startSynchronizing == null) throw new ArgumentNullException(nameof(startSynchronizing), $"{nameof(startSynchronizing)} is null.");
     if (scheduler == null) throw new ArgumentNullException(nameof(scheduler));
     _scheduler = scheduler;
     _startSynchronizing = startSynchronizing;
     _transcodingNotifications = transcodingNotifications;
     _enabledDisposable = new ReplaySubject<IDisposable>(1, _scheduler);
     _enabledDisposable.OnNext(null);
     _disposable = _enabledDisposable.Delta((d1, d2) =>
     {
         if (d1 != null)
         {                    
             d1.Dispose();
         }
         return d1 != null || d2 != null;
     })
         .TakeWhile(b => b)
         .SubscribeOn(_scheduler)
         .Subscribe(_ => { }, e => { });
 }

        public void Last_FeedItem_Is_The_One_Published_Later()
        {
            var testPodcastsSubj = new ReplaySubject<IPodcastItem>();
            IPodcastItemsLoader testPodcasts = new TestPodcastItemsLoader(testPodcastsSubj);

            var initialDate = DateTime.Now;
            testPodcastsSubj.OnNext(new TestPodcastItem(1, initialDate.AddDays(1)));
            testPodcastsSubj.OnNext(new TestPodcastItem(2, initialDate.AddDays(2)));

            _virtualScheduler.AdvanceBy(TimeSpan.FromSeconds(1));
            var model = new FeedViewModel("TestFeed", testPodcasts);
            _virtualScheduler.AdvanceBy(TimeSpan.FromSeconds(1));

            Assert.AreEqual(2, ((TestPodcastItem)model.LastFeedItem).Id);
            Assert.AreEqual(2, model.Items.Count);

            testPodcastsSubj.OnNext(new TestPodcastItem(3, initialDate.AddDays(3)));
            _virtualScheduler.AdvanceBy(TimeSpan.FromSeconds(1));

            Assert.AreEqual(3, ((TestPodcastItem)model.LastFeedItem).Id);

            testPodcastsSubj.OnNext(new TestPodcastItem(4, initialDate.AddDays(-1)));
            _virtualScheduler.AdvanceBy(TimeSpan.FromSeconds(1));
            
            Assert.AreEqual(3, ((TestPodcastItem)model.LastFeedItem).Id);
        }

        public Steps(ServiceContext serviceContext, ClientContext clientContext)
        {
            _serviceContext = serviceContext;
            _clientContext = clientContext;

            _messages = new ReplaySubject<Tuple<string, IMessage>>();
        }

 public CommandVersioningTests()
 {
     var events = new ReplaySubject<object>();
     events.Subscribe(_publishedEvents.Add);
     _host = new CedarHost(new Bootstrapper(new ObservableMessagePublisher(events)));
     _client = _host.CreateClient();
 }

        public async Task GetProgressFromASyncOperation()
        {
            CredentialsHandler credentials = (url, usernameFromUrl, types) =>
                new UsernamePasswordCredentials
                {
                    Username = "shiftkey-tester",
                    Password = "haha-password"
                };

            var repository = new ObservableRepository(
                @"C:\Users\brendanforster\Documents\GìtHūb\testing-pushspecs",
                credentials);

            Func<int, int> translate = x => x / 3;

            var pullObserver = new ReplaySubject<Tuple<string, int>>();
            var pushObserver = new ReplaySubject<Tuple<string, int>>();

            var pullResult = await repository.Pull(pullObserver);

            Assert.NotEqual(MergeStatus.Conflicts, pullResult.Status);

            await repository.Push(pushObserver);

            var list = await pullObserver.Select(x => translate(x.Item2) * 2)
                                 .Concat(pushObserver.Select(x => 67 + translate(x.Item2)))
                                 .ToList();

            Assert.NotEmpty(list);
            Assert.Equal(100, list.Last());
        }

 public LeakyLetterRepo()
 {
     _letters = new ReplaySubject<string>();
     _letters.OnNext("A");
     _letters.OnNext("B");
     _letters.OnNext("C");
 }

 ///<summary>
 ///ReplaySubject<T> will listen to all publications once subscribed. 
 ///The subscriber will also get all publications made before subscription.
 ///Simply, ReplaySubject has a buffer in whihc it will keep all the publications made for future subscriptions.
 ///</summary>
 private static void ReplaySubject()
 {
     var subject = new ReplaySubject<string>();
         subject.OnNext("a");
         subject.Subscribe(Console.WriteLine);
         subject.OnNext("b");
         subject.OnNext("c");
 }

 static IObservable<string> UsingSubject()
 {
     var subject = new ReplaySubject<string>();
       subject.OnNext("a");
       subject.OnNext("b");
       subject.OnCompleted();
       Task.Delay(TimeSpan.FromMilliseconds(3000)).Wait();
       return subject;
 }

        internal RedisLogger(string key, ILog log, IRedisConnectionFactory redisConnectionFactory)
        {
            this.key = string.Format(CultureInfo.InvariantCulture, "{0}:{1}", log.Logger.Name, key);
            this.log = log;
            this.messagesSubject = new ReplaySubject<Tuple<string, string>>(100, TimeSpan.FromSeconds(5));
            this.retry = new BehaviorSubject<bool>(false);

            var redisOnConnectionAction = new Action<Task<RedisConnection>>(task =>
            {
                if (task.IsCompleted && !task.IsFaulted)
                {
                    Interlocked.CompareExchange<RedisConnection>(ref this.redisConnection, task.Result, null);
                    subscription = messagesSubject.TakeUntil(retry.Skip(1)).Subscribe((item) => 
                        {
                            redisConnection.Publish(item.Item1, item.Item2).ContinueWith(taskWithException =>
                                {
                                    taskWithException.Exception.Handle(ex => true);
                                }, TaskContinuationOptions.OnlyOnFaulted);
                        });
                }
            });

            var redisOnErrorAction = new Action<ErrorEventArgs>(ex =>
                {
                    if (ex.IsFatal)
                    {
                        retry.OnNext(true);
                        Interlocked.Exchange<RedisConnection>(ref this.redisConnection, null);
                    }
                });

            Action subscribeAction = () =>
            {
                var connectionTask = redisConnectionFactory.CreateRedisConnection();
                connectionTask.ContinueWith(taskConnection =>
                    {
                        if (!taskConnection.IsFaulted)
                        {
                            taskConnection.ContinueWith(redisOnConnectionAction);
                            taskConnection.Result.Error += (_, err) => redisOnErrorAction(err);
                        }
                        else
                        {
                            taskConnection.Exception.Handle(_ => true);
                            this.retry.OnNext(true);
                        }
                    });
            };

            retry.Subscribe(val =>
                {
                    if (val)
                        Observable.Timer(TimeSpan.FromSeconds(10)).Subscribe(_ => subscribeAction()); 
                    else
                        subscribeAction();
                });
        }

 ///<summary>
 ///The Method simulates a blocking call by assigning the Immediate Thread as the Thread of execution
 /// The execution will move into asyncmode when we call the Nonblocking Method
 /// </summary>
 private static IObservable<string> BlockingMethod()
 {
     var subject = new ReplaySubject<string>(Scheduler.Immediate);
       subject.Subscribe(Console.WriteLine);
       subject.OnNext("a");
       subject.OnNext("b");
       subject.OnCompleted();
       Thread.Sleep(2000);
       return subject;
 }

 public static void ReplaySubjectBufferExample()
 {
     var bufferSize = 2;
     var subject = new ReplaySubject<string>(bufferSize);
     subject.OnNext("a");
     subject.OnNext("b");
     subject.OnNext("c");
     subject.Subscribe(Console.WriteLine);
     subject.OnNext("d");
 }

        public async Task EmitPointerStateWhenUpdated()
        {
            var points = new ReplaySubject<PointerState>();
            var observer = new PointerObserver(points);

            observer.Update(new GameTime(), Point.Zero);
            Assert.AreEqual(new PointerState { Position = new GeoPoint() }, await points.FirstAsync());

            observer.Update(new GameTime(), new Point(Config.SpriteSize, Config.SpriteSize));
            Assert.AreEqual(new PointerState { Position = new GeoPoint { X = 1, Y = 1 } }, await points.Skip(1).FirstAsync());
        }

 public static void ReplaySubjectWindowExample()
 {
     var window = TimeSpan.FromMilliseconds(150);
     var subject = new ReplaySubject<string>(window);
     subject.OnNext("w");
     Thread.Sleep(TimeSpan.FromMilliseconds(100));
     subject.OnNext("x");
     Thread.Sleep(TimeSpan.FromMilliseconds(100));
     subject.OnNext("y");
     subject.Subscribe(Console.WriteLine);
     subject.OnNext("z");
 }

        public static async Task Returns_Correct_Movement(decimal price1, decimal price2, PriceMovement expected)
        {
            var subject = new ReplaySubject<IPrice>();

            var result = subject.ToPriceMovementStream();
            subject.OnNext(new Price { Mid = price1 });
            subject.OnNext(new Price { Mid = price2 });
            subject.OnCompleted();

            // Assert
            await result.SingleAsync(movement => movement == expected);
        }

        private void Start()
        {
            //ReplaySubjectも初期を設定することができる（今回は設定しない）
            replaySubject = new ReplaySubject<int>();

            /*
            OnNextを何回か繰り返す → Subscribe と実行し、Subscribeした瞬間に過去の値がまとめて値が発行されることを確認しよう
            */

            // Subscribeボタンが押されたらSubjectをSubscribeしてresultTextに表示する
            subscribeButton.OnClickAsObservable().Subscribe(_ =>
            {
                if (replaySubject != null)
                {
                    replaySubject.Subscribe(
                        time => resultText.text += time.ToString() + "　", //OnNext
                        () => resultText.text += "OnCompleted　"); //OnCompleted
                }
            });

            // OnNextボタンが押されたら今が何度目のOnNextであるかを発行する
            onNextButton.OnClickAsObservable().Subscribe(_ =>
            {
                if (replaySubject != null)
                {
                    replaySubject.OnNext(++onNextCount);
                }
            });

            // OnCompletedボタンが押されたらOnCompletedを発行する
            onCompletedButton.OnClickAsObservable().Subscribe(_ =>
            {
                if (replaySubject != null)
                {
                    replaySubject.OnCompleted();
                }
            });

            // Resetボタンが押されたら全体を初期化する
            resetButton.OnClickAsObservable().Subscribe(_ =>
            {
                if (replaySubject != null)
                {
                    replaySubject.OnCompleted();
                }
                replaySubject = new ReplaySubject<int>();
                resultText.text = "";
                onNextCount = 0;
            });
        }

        public void ApplyReleasesWithDeltaReleases()
        {
            string tempDir;

            using (Utility.WithTempDirectory(out tempDir)) {
                Directory.CreateDirectory(Path.Combine(tempDir, "theApp", "packages"));

                new[] {
                    "Shimmer.Core.1.0.0.0-full.nupkg",
                    "Shimmer.Core.1.1.0.0-delta.nupkg",
                    "Shimmer.Core.1.1.0.0-full.nupkg",
                }.ForEach(x => File.Copy(IntegrationTestHelper.GetPath("fixtures", x), Path.Combine(tempDir, "theApp", "packages", x)));

                var fixture = new UpdateManager("http://lol", "theApp", FrameworkVersion.Net40, tempDir, null, new FakeUrlDownloader());

                var baseEntry = ReleaseEntry.GenerateFromFile(Path.Combine(tempDir, "theApp", "packages", "Shimmer.Core.1.0.0.0-full.nupkg"));
                var deltaEntry = ReleaseEntry.GenerateFromFile(Path.Combine(tempDir, "theApp", "packages", "Shimmer.Core.1.1.0.0-delta.nupkg"));
                var latestFullEntry = ReleaseEntry.GenerateFromFile(Path.Combine(tempDir, "theApp", "packages", "Shimmer.Core.1.1.0.0-full.nupkg"));

                var updateInfo = UpdateInfo.Create(baseEntry, new[] { deltaEntry, latestFullEntry }, "dontcare", FrameworkVersion.Net40);
                updateInfo.ReleasesToApply.Contains(deltaEntry).ShouldBeTrue();

                using (fixture) {
                    var progress = new ReplaySubject<int>();

                    fixture.ApplyReleases(updateInfo, progress).First();
                    this.Log().Info("Progress: [{0}]", String.Join(",", progress));

                    progress.Buffer(2,1).All(x => x.Count != 2 || x[1] > x[0]).First().ShouldBeTrue();
                    progress.Last().ShouldEqual(100);
                }

                var filesToFind = new[] {
                    new {Name = "NLog.dll", Version = new Version("2.0.0.0")},
                    new {Name = "NSync.Core.dll", Version = new Version("1.1.0.0")},
                    new {Name = "Ionic.Zip.dll", Version = new Version("1.9.1.8")},
                };

                filesToFind.ForEach(x => {
                    var path = Path.Combine(tempDir, "theApp", "app-1.1.0.0", x.Name);
                    this.Log().Info("Looking for {0}", path);
                    File.Exists(path).ShouldBeTrue();

                    var vi = FileVersionInfo.GetVersionInfo(path);
                    var verInfo = new Version(vi.FileVersion ?? "1.0.0.0");
                    x.Version.ShouldEqual(verInfo);
                });
            }
        }

            public async Task fires_once_when_a_file_is_changed()
            {
                // Given
                File.WriteAllText(fileName, "hubba");
                var results = new ReplaySubject<string>();
                FileSystem.WhenFileChanges(fileName).Subscribe(results);

                // When
                File.WriteAllText(fileName, "bubba");

                // Then
                var events = await results.TakeUntil(DateTime.Now + TimeSpan.FromMilliseconds(200)).ToArray();
                events.Should().HaveCount(1);
                events[0].Should().Be(fileName);
            }

        public void SubscribeBeforeSetParam()
        {
            var server = new ParameterServer(new Uri("http://localhost"));

            var slave = new SlaveServer("test", 0, new TopicContainer());
            var observer = new ReplaySubject<KeyValuePair<string, object>>();

            slave.ParameterUpdated += (key, value) => observer.OnNext(new KeyValuePair<string, object>(key, value));

            server.SubscribeParam("test", slave.SlaveUri.ToString(), "test_param");

            server.SetParam("test", "test_param", 5678);

            observer.First().Key.Is("test_param");
            observer.First().Value.Is(5678);
        }

        public void honours_buffer_size_for_replays_with_priority_to_most_recent()
        {
            ReplaySubject<int> subject = new ReplaySubject<int>(2);

            StatsObserver<int> stats = new StatsObserver<int>();

            subject.OnNext(1);
            subject.OnNext(2);
            subject.OnNext(3);

            subject.Subscribe(stats);

            Assert.AreEqual(2, stats.NextCount);
            Assert.IsTrue(stats.NextValues.SequenceEqual(new int[] { 2, 3 }));
            Assert.IsFalse(stats.CompletedCalled);
        }