public void live_values_are_sent_through_scheduler()
        {
            ManualScheduler scheduler = new ManualScheduler();

            BehaviorSubject<int> subject = new BehaviorSubject<int>(0, scheduler);

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

            subject.Subscribe(stats);

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

            Assert.IsFalse(stats.NextCalled);

            scheduler.RunNext();
            Assert.AreEqual(1, stats.NextCount);
            Assert.IsTrue(stats.NextValues.SequenceEqual(new int[] { 0 }));
            Assert.IsFalse(stats.CompletedCalled);

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

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

            scheduler.RunNext();
            Assert.IsTrue(stats.CompletedCalled);
        }

 public SharedThemeService(ResourceDictionary resourceDictionary)
 {
     _resourceDictionary = resourceDictionary;
     _themes = new BehaviorSubject<IEnumerable<ITheme>>(Enumerable.Empty<ITheme>());
     _activeTheme = new BehaviorSubject<ITheme>(null);
     _themesMap = new Dictionary<string, ITheme>();
 }

        public MainWindowViewModel()
        {
            var noneInFlight = new BehaviorSubject<bool>(false);
            var updateManager = default(UpdateManager);

            this.WhenAny(x => x.UpdatePath, x => x.Value)
                .Where(x => !String.IsNullOrWhiteSpace(x))
                .Throttle(TimeSpan.FromMilliseconds(700), RxApp.DeferredScheduler)
                .Subscribe(x => {
                    if (updateManager != null)  updateManager.Dispose();
                    updateManager = new UpdateManager(UpdatePath, "SampleUpdatingApp", FrameworkVersion.Net40);
                });

            CheckForUpdate = new ReactiveAsyncCommand(noneInFlight);
            CheckForUpdate.RegisterAsyncObservable(_ => updateManager.CheckForUpdate())
                .Subscribe(x => { UpdateInfo = x; DownloadedUpdateInfo = null; });

            DownloadReleases = new ReactiveAsyncCommand(noneInFlight.Where(_ => UpdateInfo != null));
            DownloadReleases.RegisterAsyncObservable(_ => updateManager.DownloadReleases(UpdateInfo.ReleasesToApply))
                .Subscribe(_ => DownloadedUpdateInfo = UpdateInfo);

            ApplyReleases = new ReactiveAsyncCommand(noneInFlight.Where(_ => DownloadedUpdateInfo != null));
            ApplyReleases.RegisterAsyncObservable(_ => updateManager.ApplyReleases(DownloadedUpdateInfo));

            Observable.CombineLatest(
                CheckForUpdate.ItemsInflight.StartWith(0),
                DownloadReleases.ItemsInflight.StartWith(0),
                ApplyReleases.ItemsInflight.StartWith(0),
                this.WhenAny(x => x.UpdatePath, _ => 0),
                (a, b, c, _) => a + b + c
            ).Select(x => x == 0 && UpdatePath != null).Multicast(noneInFlight).Connect();
        }

        public void Should_Produce_Correct_Values()
        {
            var activator = new BehaviorSubject<bool>(false);
            var source = new BehaviorSubject<object>(1);
            var target = new ActivatedObservable(activator, source, string.Empty);
            var result = new List<object>();

            target.Subscribe(x => result.Add(x));

            activator.OnNext(true);
            source.OnNext(2);
            activator.OnNext(false);
            source.OnNext(3);
            activator.OnNext(true);

            Assert.Equal(
                new[] 
                {
                    AvaloniaProperty.UnsetValue,
                    1,
                    2,
                    AvaloniaProperty.UnsetValue,
                    3,
                }, 
                result);
        }

        public MainMenu(Engine engine)
            : base(engine)
        {
            this.button1 = new MenuButton(this.Engine);
            this.button2 = new MenuButton(this.Engine);

            this.button1.Position = new Vector2(0, -40);
            this.button2.Position = new Vector2(0, 40);

            this.button1.Text = "new game";
            this.button2.Text = "high scores";

            this.button1.Color = new Color(0.3f, 0.3f, 0.3f);
            this.button2.Color = new Color(0.3f, 0.3f, 0.3f);

            button1.Action = () =>
                                 {
                                     this.Dispose();
                                     new NewGameMenu(this.Engine).Initialize().Attach();
                                 };

            button2.Action = () =>
                                 {
                                     this.Dispose();
                                     new HighScoreMenu(this.Engine).Initialize().Attach();
                                 };

            this.buttons.Add(button1);
            this.buttons.Add(button2);

            this.oldButton = new BehaviorSubject<MenuButton>(null);
            this.currentButton = new BehaviorSubject<MenuButton>(this.button1);
        }

        public void OneTime_Binding_Should_Be_Set_Up()
        {
            var dataContext = new BehaviorSubject<object>(null);
            var expression = new BehaviorSubject<object>(null);
            var target = CreateTarget(dataContext: dataContext);
            var binding = new Binding
            {
                Path = "Foo",
                Mode = BindingMode.OneTime,
            };

            binding.Bind(target.Object, TextBox.TextProperty, expression);

            target.Verify(x => x.SetValue(
                (PerspexProperty)TextBox.TextProperty, 
                null, 
                BindingPriority.LocalValue));
            target.ResetCalls();

            expression.OnNext("foo");
            dataContext.OnNext(1);

            target.Verify(x => x.SetValue(
                (PerspexProperty)TextBox.TextProperty,
                "foo",
                BindingPriority.LocalValue));
        }

        public ExecutionContext(TimeSpan skipAhead = default(TimeSpan))
        {
            this.disposables = new CompositeDisposable();
            this.cancelRequested = new BehaviorSubject<bool>(false)
                .AddTo(this.disposables);
            this.progressDeltas = new Subject<TimeSpan>()
                .AddTo(this.disposables);

            this
                .cancelRequested
                .Where(x => x)
                .Subscribe(_ => this.IsCancelled = true)
                .AddTo(this.disposables);

            this
                .progressDeltas
                .Scan((running, next) => running + next)
                .Subscribe(x => this.Progress = x)
                .AddTo(this.disposables);

            this
                .WhenAnyValue(x => x.CurrentExercise)
                .Select(x => this.progressDeltas.StartWith(TimeSpan.Zero).Scan((running, next) => running + next))
                .Switch()
                .Subscribe(x => this.CurrentExerciseProgress = x)
                .AddTo(this.disposables);

            this
                .progressDeltas
                .StartWith(skipAhead)
                .Scan((running, next) => running - next)
                .Select(x => x < TimeSpan.Zero ? TimeSpan.Zero : x)
                .Subscribe(x => this.SkipAhead = x)
                .AddTo(this.disposables);
        }

        internal AudioPlayer()
        {
            this.audioPlayerCallback = new DummyMediaPlayerCallback();
            this.videoPlayerCallback = new DummyMediaPlayerCallback();
            this.currentCallback = new DummyMediaPlayerCallback();

            this.finishSubscription = new SerialDisposable();
            this.gate = new SemaphoreSlim(1, 1);

            this.playbackState = new BehaviorSubject<AudioPlayerState>(AudioPlayerState.None);
            this.PlaybackState = this.playbackState.DistinctUntilChanged();

            this.loadedSong = new BehaviorSubject<Song>(null);
            this.TotalTime = this.loadedSong.Select(x => x == null ? TimeSpan.Zero : x.Duration);

            this.currentTimeChangedFromOuter = new Subject<TimeSpan>();

            var conn = Observable.Interval(TimeSpan.FromMilliseconds(300), RxApp.TaskpoolScheduler)
                .CombineLatest(this.PlaybackState, (l, state) => state)
                .Where(x => x == AudioPlayerState.Playing)
                .Select(_ => this.CurrentTime)
                .Merge(this.currentTimeChangedFromOuter)
                .DistinctUntilChanged(x => x.TotalSeconds)
                .Publish(TimeSpan.Zero);
            conn.Connect();
            this.CurrentTimeChanged = conn;
        }

        public AnalyticsEngine()
        {
            _currentPositionUpdatesDto = new PositionUpdatesDto();
            _updates = new BehaviorSubject<PositionUpdatesDto>(_currentPositionUpdatesDto);

            _eventLoopScheduler.SchedulePeriodic(TimeSpan.FromSeconds(10), PublishPositionReport);
        }

        public async void Should_Produce_Value_On_Activator_True()
        {
            var activator = new BehaviorSubject<bool>(true);
            var target = new StyleBinding(activator, 1, string.Empty);
            var result = await target.Take(1);

            Assert.Equal(1, result);
        }

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

 protected OnOffFeature(bool @on)
 {
     availability = new BehaviorSubject<bool>(@on);
     activator = new FeatureActivator(
         Activate,
         Deactivate,
         availability);
 }

        public iCloudExerciseDocumentService(ILoggerService loggerService)
        {
            Ensure.ArgumentNotNull(loggerService, nameof(loggerService));

            this.logger = loggerService.GetLogger(this.GetType());
            this.exerciseDocument = new BehaviorSubject<string>(null);
            this.sync = new object();
        }

        public async void Should_Produce_UnsetValue_On_Activator_False()
        {
            var activator = new BehaviorSubject<bool>(false);
            var target = new StyleBinding(activator, 1, string.Empty);
            var result = await target.Take(1);

            Assert.Equal(PerspexProperty.UnsetValue, result);
        }

        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();
                });
        }

        public ServiceControl(IBlobCache cache, string url = null)
        {
            this.cache = cache;
            subject = new BehaviorSubject<Unit>(Unit.Default);
            isValid = new BehaviorSubject<bool>(false);
            IsValid = isValid.AsObservable();

            UpdateUrl(url ?? "http://localhost:33333/api").Wait();
        }

 static void Example4()
 {
     var subject = new BehaviorSubject<string>("a");
       subject.OnNext("b");
       subject.OnNext("c");
       subject.OnNext("d");
       subject.OnCompleted();
       subject.Subscribe(Console.WriteLine);
 }

    protected override void Main()
    {
      if (Environment.OSVersion.Version < new Version(6, 2))
      {
        TraceError(Text.LabRequiresWindows8OrHigher);
        return;
      }

      const int port = 5494;
      string subProtocol = GetType().Name;

      var userMessages = new BehaviorSubject<string>(null);

      var client = new ClientWebSocket();

      client.Options.AddSubProtocol(subProtocol);

      using (client)
      using (var cancel = new CancellationDisposable())
      using (ObservableHttpListener
        .StartWebSockets(new IPEndPoint(IPAddress.Loopback, port), subProtocol)
        .Subscribe(
          async request =>
          {
            using (var socket = request.WebSocket)
            {
              try
              {
                var message = await ReceiveMessageAsync(socket, cancel.Token);
                await SendMessageAsync("You sent \"" + message + "\"", socket, cancel.Token);
                await ReceiveCloseMessageAsync(socket, cancel.Token);
              }
              catch (OperationCanceledException)
              {
              }
            }
          },
          TraceError))
      using ((from _ in client.ConnectAsync(new Uri("ws://localhost:" + port), cancel.Token).ToObservable()
                .Do(_ => TraceLine(Environment.NewLine + "(Connected to host on sub-protocol \"{0}\")", client.SubProtocol))
              from message in userMessages.Where(m => m != null).Take(1)
              from __ in SendMessageAsync(message, client, cancel.Token).ToObservable()
              from response in ReceiveMessageAsync(client, cancel.Token).ToObservable()
                .Do(___ => client.CloseAsync(WebSocketCloseStatus.NormalClosure, "Done", cancel.Token))
              select response)
              .Subscribe(
                response => TraceLine("Response: {0}", response),
                TraceError,
                () => TraceLine("{0}: {1}", Text.Client, Text.Done)))
      {
        userMessages.OnNext(UserInput("{0}> ", Instructions.EnterAMessage));

        TraceStatus(Instructions.PressAnyKeyToCancel);

        WaitForKey();
      }
    }

        /// <summary>
        /// Initializes a new instance of the <see cref="LocalSong" /> class.
        /// </summary>
        /// <param name="path">The path of the file.</param>
        /// <param name="duration">The duration of the song.</param>
        /// <param name="artworkKey">
        /// The key of the artwork for Akavache to retrieve. Null, if there is no album cover or the
        /// artwork isn't retrieved yet.
        /// </param>
        public LocalSong(string path, TimeSpan duration, string artworkKey = null)
            : base(path, duration)
        {
            if (artworkKey == String.Empty)
                Throw.ArgumentException("Artwork key cannot be an empty string", () => artworkKey);

            this.artworkKey = new BehaviorSubject<string>(artworkKey);

            this.Guid = Guid.NewGuid();
        }

 ///<summary>
 ///With BehaviourSubject<T> ,the subscriber will only get all the last publication made
 ///Simply, BehaviourSubject has a one value buffer. Hence, it requires a default value.
 ///</summary>
 private static void BehaviourSubject()
 {
     var subject = new BehaviorSubject<string>("Rx");
         subject.OnNext("a");
         var d = subject.Subscribe(x => Console.WriteLine("Subscritipon 1 : " + x));
         subject.OnNext("b");
        // var d = subject.Subscribe(x => Console.WriteLine("Subscritipon 1 : " + x));
         d.Dispose();
         subject.OnNext("c");
         subject.Subscribe(x => Console.WriteLine("Subscritipon 2 : " + x));
 }