public void SetUp()
        {
            disposables = new CompositeDisposable
            {
                VirtualClock.Start()
            };

            clockName = Any.CamelCaseName();
            targetId = Any.Word();
            target = new CommandTarget(targetId);
            store = new InMemoryStore<CommandTarget>(
                _ => _.Id,
                id => new CommandTarget(id))
            {
                target
            };

            CommandSchedulerDbContext.NameOrConnectionString =
                @"Data Source=(localdb)\MSSQLLocalDB; Integrated Security=True; MultipleActiveResultSets=False; Initial Catalog=ItsCqrsTestsCommandScheduler";

            configuration = new Configuration()
                .UseInMemoryCommandScheduling()
                .UseDependency<IStore<CommandTarget>>(_ => store)
                .UseDependency<GetClockName>(c => _ => clockName)
                .TraceScheduledCommands();

            scheduler = configuration.CommandScheduler<CommandTarget>();

            Command<CommandTarget>.AuthorizeDefault = (commandTarget, command) => true;

            disposables.Add(ConfigurationContext.Establish(configuration));
            disposables.Add(configuration);
        }

        public IDisposable Activate()
        {
            var disp = new CompositeDisposable(blocks.SelectMany(x => x()));

            Interlocked.Exchange(ref activationHandle, disp).Dispose();
            return Disposable.Create(Deactivate);
        }

        public void SetUp()
        {
            eventStoreDbTest = new EventStoreDbTest();
            clockName = Any.CamelCaseName();

            Clock.Reset();

            disposables = new CompositeDisposable
            {
                Disposable.Create(() => eventStoreDbTest.TearDown()),
                Disposable.Create(Clock.Reset)
            };

            var bus = new FakeEventBus();
            orderRepository = new SqlEventSourcedRepository<Order>(bus);
            accountRepository = new SqlEventSourcedRepository<CustomerAccount>(bus);

            var configuration = new Configuration();
            configuration.UseEventBus(bus)
                         .UseDependency<IEventSourcedRepository<Order>>(t => orderRepository)
                         .UseDependency<IEventSourcedRepository<CustomerAccount>>(t => accountRepository);

            ConfigureScheduler(configuration);

            disposables.Add(ConfigurationContext.Establish(configuration));

            Console.WriteLine(new { clockName });

            clockTrigger = configuration.Container.Resolve<ISchedulerClockTrigger>();
            clockRepository = configuration.Container.Resolve<ISchedulerClockRepository>();
            clockRepository.CreateClock(clockName, Clock.Now());
        }

    public static void Run()
    {
      var systems = new[] { "A", "B", "C" }.Select(CreateExternalSystem);
      var observables = systems.Select(s => s.ObserveHealth()).Select(obs => obs.DistinctUntilChanged(c => c.IsAvailable)).ToList();
      var disposable = new CompositeDisposable();

      // observe independently
      disposable.Add(new CompositeDisposable(observables.Select(c => c.Subscribe(PrintHealthCheck))));

      // merge
      var merged = observables.Aggregate((l, r) => l.Merge(r));
      disposable.Add(merged.Subscribe(PrintHealthCheck));

      // combine
      var combined = observables
        .Aggregate(Observable.Return(Enumerable.Empty<HealthCheck>()), (agg, obs) => agg.CombineLatest(obs, (checks, check) => checks.Concat(new[] { check })));

      var scan = merged.Scan(ImmutableDictionary<string, bool>.Empty, (d, check) => d.SetItem(check.ExternalSystemName, check.IsAvailable));
      
      disposable.Add(combined.Subscribe(e => Console.WriteLine("Combined: " + string.Join(", ", e.Select(c => $"{c.ExternalSystemName}={c.IsAvailable}")))));
      disposable.Add(scan.Subscribe(d => Console.WriteLine("Scanned: " + string.Join(", ", d.Select(p => $"{p.Key}={p.Value}")))));

      Console.ReadKey();
      disposable.Dispose();
    }

        public IObservable<Unit> SendMessage(string message, IScheduler scheduler)
        {
            return Observable.Create<Unit>(observer =>
            {
                var disposable = new CompositeDisposable();
                var buffer = Encoding.UTF8.GetBytes(message);
                connectionToken.SocketEvent.SetBuffer(buffer, 0, buffer.Length);

                var disposableCompletedSubscription = connectionToken.SocketEvent.Completed.Subscribe(_ =>
                {
                    SendNotificationToObserver(observer, connectionToken.SocketEvent);
                });

                var disposableActions = scheduler.Schedule(() =>
                {
                    if (!connectionToken.Socket.SendAsync(connectionToken.SocketEvent))
                    {
                        SendNotificationToObserver(observer, connectionToken.SocketEvent);
                    }
                });

                disposable.Add(disposableCompletedSubscription);
                disposable.Add(disposableActions);

                return disposable;
            });
        }

        public IDisposable Subscribe(ReactiveSpace spaceListener)
        {
            CompositeDisposable subscriptions = new CompositeDisposable();
            subscriptions.Add(spaceListener
                                .LockedHands()
                                .ObserveOn(UI)
                                .Subscribe(o =>
                                {
                                    HandsCount++;
                                }));

            subscriptions.Add(spaceListener
                                .LockedHands()
                                .Select(o =>
                                        o
                                        .ObserveOn(UI)
                                        .Subscribe(oo =>
                                        {
                                        }, () =>
                                        {
                                            HandsCount--;
                                        }))
                                .Subscribe());
            subscriptions.Add(SubscribeCore(spaceListener));
            subscriptions.Add(Disposable.Create(()=>HandsCount = 0));
            return subscriptions;
        }

        public App()
        {
            var dir = System.AppDomain.CurrentDomain.BaseDirectory;

            this.WindowPlacement = new WindowPlace(dir + @"placement.config");
            this.disposables = new CompositeDisposable();
        }

        public void SetUp()
        {
            disposables = new CompositeDisposable();
            telemetryEvents = new List<Telemetry>();

            disposables.Add(Log.TelemetryEvents().Subscribe(e => { telemetryEvents.Add(e); }));
        }

        public void SetUp()
        {
            disposables = new CompositeDisposable
            {
                VirtualClock.Start()
            };

            clockName = Any.CamelCaseName();
            targetId = Any.Word();
            target = new CommandTarget(targetId);
            store = new InMemoryStore<CommandTarget>(
                _ => _.Id,
                id => new CommandTarget(id))
            {
                target
            };

            configuration = new Configuration()
                .UseInMemoryCommandScheduling()
                .UseDependency<IStore<CommandTarget>>(_ => store)
                .UseDependency<GetClockName>(c => _ => clockName)
                .TraceScheduledCommands();

            scheduler = configuration.CommandScheduler<CommandTarget>();

            Command<CommandTarget>.AuthorizeDefault = (commandTarget, command) => true;

            disposables.Add(ConfigurationContext.Establish(configuration));
            disposables.Add(configuration);
        }

        protected virtual IDisposable LockChildren(LockTypes lockType)
        {
            var result = new CompositeDisposable();

            lock (ChildLocksSync)
            {
                foreach (var child in ChildLocks)
                {
                    if (lockType == LockTypes.Read)
                    {
                        result.AddIfNotNull(child.AcquireReadLockIfNotHeld());
                    }
                    else if (lockType == LockTypes.UpgradeableRead)
                    {
                        result.AddIfNotNull(child.AcquireUpgradeableReadLock());
                    }
                    else if (lockType == LockTypes.Write)
                    {
                        result.AddIfNotNull(child.AcquireWriteLockIfNotHeld());
                    }
                    else
                    {
                        throw new NotSupportedException(lockType.ToString());
                    }
                }
            }

            return result;
        }

        protected override IDisposable SubscribeCore(ReactiveSpace spaceListener)
        {
            CompositeDisposable subscriptions = new CompositeDisposable();

            subscriptions.Add(spaceListener
                .LockedHands()
                .ObserveOn(UI)
                .SelectMany(h => h
                                .Select(hh => new
                                {
                                    Group = h,
                                    Hand = hh
                                }))
                .Subscribe(h =>
                {
                    var diff = 1000 + (h.Hand.PalmPosition.y - h.Group.Key.PalmPosition.y);
                    var bin = (int)(diff / MinInterval);
                    if(bin < PreviousBin)
                    {
                        if(OnMoveDown != null)
                            OnMoveDown();
                    }
                    if(bin > PreviousBin)
                    {
                        if(OnMoveUp != null)
                            OnMoveUp();
                    }
                    PreviousBin = bin;
                }));

            return subscriptions;
        }

        public void SelectDisposableShouldWork()
        {
            var scheduler = new TestScheduler();
            var disposables = new List<BooleanDisposable>();
            var list = new CompositeDisposable();
            scheduler.CreateColdObservable(
                new Recorded<Notification<long>>(100, Notification.CreateOnNext(0L)),
                new Recorded<Notification<long>>(200, Notification.CreateOnNext(1L)),
                new Recorded<Notification<long>>(300, Notification.CreateOnNext(2L)),
                new Recorded<Notification<long>>(400, Notification.CreateOnNext(3L)),
                new Recorded<Notification<long>>(400, Notification.CreateOnCompleted<long>())
            )
            .SelectDisposable(list, i => {
                var d = new BooleanDisposable();
                disposables.Add(d);
                return d;
            }, (i, _) => i)
            .Subscribe()
            .DisposeWith(list);

            scheduler.AdvanceTo(300);

            disposables.Count.Should().Be(3);

            disposables.Select(d => d.IsDisposed).Should().NotContain(true);

            list.Dispose();

            disposables.Select(d => d.IsDisposed).Should().NotContain(false);

        }

 /// <param name="type">The UIViewType</param>
 /// <param name="v">The IView</param>
 /// <param name="vm">The IViewModel. Might be null because the 2fa view shares the same viewmodel as the login dialog, so it's
 /// set manually in the view outside of this</param>
 public UIPair(UIViewType type, ExportLifetimeContext<IView> v, [AllowNull]ExportLifetimeContext<IViewModel> vm)
 {
     viewType = type;
     view = v;
     viewModel = vm;
     handlers = new CompositeDisposable();
 }

        public void SetUp()
        {
            aggregateId = Any.Guid();
            sequenceNumber = Any.PositiveInt();

            disposables = new CompositeDisposable
            {
                ConfigurationContext.Establish(new Configuration()
                                                   .UseSqlStorageForScheduledCommands(c => c.UseConnectionString(TestDatabases.CommandScheduler.ConnectionString)))
            };

            if (clockName == null)
            {
                clockName = Any.CamelCaseName();

                using (var db = Configuration.Current.CommandSchedulerDbContext())
                {
                    db.Clocks.Add(new CommandScheduler.Clock
                    {
                        Name = clockName,
                        StartTime = Clock.Now(),
                        UtcNow = Clock.Now()
                    });

                    db.SaveChanges();
                }
            }

            
            using (var db = Configuration.Current.CommandSchedulerDbContext())
            {
                db.Database.ExecuteSqlCommand("delete from PocketMigrator.AppliedMigrations where MigrationScope = 'CommandSchedulerCleanup'");
            }
        }

        public void SetUp()
        {
            clockName = Any.CamelCaseName();

            Clock.Reset();

            disposables = new CompositeDisposable
            {
                Disposable.Create(Clock.Reset)
            };

            var configuration = new Configuration()
                .UseSqlEventStore(c => c.UseConnectionString(TestDatabases.EventStore.ConnectionString))
                .UseSqlStorageForScheduledCommands(c => c.UseConnectionString(TestDatabases.CommandScheduler.ConnectionString));

            Configure(configuration);

            disposables.Add(ConfigurationContext.Establish(configuration));
            disposables.Add(configuration);

            orderRepository = configuration.Repository<Order>();
            accountRepository = configuration.Repository<CustomerAccount>();
            clockTrigger = configuration.SchedulerClockTrigger();
            clockRepository = configuration.SchedulerClockRepository();
            clockRepository.CreateClock(clockName, Clock.Now());
        }

        public KanColleProxy()
        {
            this.compositeDisposable = new CompositeDisposable();

            this.connectableSessionSource = Observable
                .FromEvent<Action<Session>, Session>(
                    action => action,
                    h => HttpProxy.AfterSessionComplete += h,
                    h => HttpProxy.AfterSessionComplete -= h)
                .Publish();

            this.apiSource = this.connectableSessionSource
                .Where(s => s.Request.PathAndQuery.StartsWith("/kcsapi"))
                .Where(s => s.Response.MimeType.Equals("text/plain"))
                #region .Do(debug)
            #if DEBUG
            .Do(session =>
                {
                    Debug.WriteLine("==================================================");
                    Debug.WriteLine("Nekoxy session: ");
                    Debug.WriteLine(session);
                    Debug.WriteLine("");
                })
            #endif
            #endregion
                .Publish();
        }

        /// <summary>
        /// Initializes a new instance of the <see cref="PlaylistViewModel" /> class.
        /// </summary>
        /// <param name="playlist">The playlist info.</param>
        /// <param name="renameRequest">
        /// A function that requests the rename of the playlist. Return true, if the rename is
        /// granted, otherwise false.
        /// </param>
        public PlaylistViewModel(Playlist playlist, Func<string, bool> renameRequest)
        {
            this.playlist = playlist;
            this.renameRequest = renameRequest;

            this.disposable = new CompositeDisposable();

            this.entries = playlist
                .CreateDerivedCollection(entry => new PlaylistEntryViewModel(entry))
                .DisposeWith(this.disposable);
            this.entries.ItemsRemoved.Subscribe(x => x.Dispose());

            this.playlist.WhenAnyValue(x => x.CurrentSongIndex).ToUnit()
                .Merge(this.entries.Changed.ToUnit())
                .Subscribe(_ => this.UpdateCurrentSong())
                .DisposeWith(this.disposable);

            IObservable<List<PlaylistEntryViewModel>> remainingSongs = this.entries.Changed
                .Select(x => Unit.Default)
                .Merge(this.playlist.WhenAnyValue(x => x.CurrentSongIndex).ToUnit())
                .Select(x => this.entries.Reverse().TakeWhile(entry => !entry.IsPlaying).ToList());

            this.songsRemaining = remainingSongs
                .Select(x => x.Count)
                .ToProperty(this, x => x.SongsRemaining)
                .DisposeWith(this.disposable);

            this.timeRemaining = remainingSongs
                .Select(x => x.Any() ? x.Select(entry => entry.Duration).Aggregate((t1, t2) => t1 + t2) : (TimeSpan?)null)
                .ToProperty(this, x => x.TimeRemaining)
                .DisposeWith(this.disposable);

            this.CurrentPlayingEntry = this.Model.WhenAnyValue(x => x.CurrentSongIndex).Select(x => x == null ? null : this.entries[x.Value]);
        }

        public void SetUp()
        {
            // disable authorization
            Command<Order>.AuthorizeDefault = (o, c) => true;
            Command<CustomerAccount>.AuthorizeDefault = (o, c) => true;

            disposables = new CompositeDisposable
            {
                VirtualClock.Start()
            };

            customerAccountId = Any.Guid();

            configuration = new Configuration()
                .UseInMemoryCommandScheduling()
                .UseInMemoryEventStore();

            customerRepository = configuration.Repository<CustomerAccount>();
            orderRepository = configuration.Repository<Order>();

            customerRepository.Save(new CustomerAccount(customerAccountId).Apply(new ChangeEmailAddress(Any.Email())));

            disposables.Add(ConfigurationContext.Establish(configuration));
            disposables.Add(configuration);
        }

        public void SetUp()
        {
            ints = Enumerable.Range(1, 1000).ToArray();
            disposables = new CompositeDisposable();
            partitionedStream = Stream
                .Partitioned<int, int, int>(
                    query: async (q, p) =>
                    {
                        return ints
                            .Where(i => i.IsWithinPartition(p))
                            .Skip(q.Cursor.Position)
                            .Take(q.BatchSize.Value);
                    },
                    advanceCursor: (query, batch) =>
                    {
                        // putting the cursor and the partition on the same field is a little weird because a batch of zero doesn't necessarily signify the end of the batch
                        if (batch.Any())
                        {
                            query.Cursor.AdvanceTo(batch.Last());
                        }
                    });

            Formatter.ListExpansionLimit = 100;
            Formatter<Projection<HashSet<int>, int>>.RegisterForAllMembers();
        }

        public void TestServiceBusWithEmbeddedBroker()
        {
            // use the embedded broker
            var brokerUri = _broker.FailoverUri;

            // set up ServiceBus using fluent interfaces and all current endpoints and pointing at test AMQ broker
            IServiceBus serviceBus = ServiceBus.Configure()
                .WithActiveMQEndpoints<ITestMessage1>()
                    .Named("Obvs.TestService")
                    .UsingQueueFor<TestCommand>().ClientAcknowledge()
                    .UsingQueueFor<TestCommand2>().ClientAcknowledge()
                    .UsingQueueFor<IRequest>().AutoAcknowledge()
                    .ConnectToBroker(brokerUri)
                    .SerializedAsJson()
                    .AsClientAndServer()
                .PublishLocally()
                    .OnlyMessagesWithNoEndpoints()
                .UsingConsoleLogging()
                .Create();

            // create threadsafe collection to hold received messages in
            ConcurrentBag<IMessage> messages = new ConcurrentBag<IMessage>();

            // create some actions that will act as a fake services acting on incoming commands and requests
            Action<TestCommand> fakeService1 = command => serviceBus.PublishAsync(new TestEvent {Id = command.Id});
            Action<TestRequest> fakeService2 = request => serviceBus.ReplyAsync(request, new TestResponse {Id = request.Id});
            AnonymousObserver<IMessage> observer = new AnonymousObserver<IMessage>(messages.Add, Console.WriteLine, () => Console.WriteLine("OnCompleted"));

            // subscribe to all messages on the ServiceBus
            CompositeDisposable subscriptions = new CompositeDisposable
            {
                serviceBus.Events.Subscribe(observer),
                serviceBus.Commands.Subscribe(observer),
                serviceBus.Requests.Subscribe(observer),
                serviceBus.Commands.OfType<TestCommand>().Subscribe(fakeService1),
                serviceBus.Requests.OfType<TestRequest>().Subscribe(fakeService2)
            };
            
            // send some messages
            serviceBus.SendAsync(new TestCommand { Id = 123 });
            serviceBus.SendAsync(new TestCommand2 { Id = 123 });
            serviceBus.SendAsync(new TestCommand3 { Id = 123 });
            serviceBus.GetResponses(new TestRequest { Id = 456 }).Subscribe(observer);

            // wait some time until we think all messages have been sent and received over AMQ
            Thread.Sleep(TimeSpan.FromSeconds(1));

            // test we got everything we expected
            Assert.That(messages.OfType<TestCommand>().Count() == 1, "TestCommand not received");
            Assert.That(messages.OfType<TestCommand2>().Count() == 1, "TestCommand2 not received");
            Assert.That(messages.OfType<TestCommand3>().Count() == 1, "TestCommand3 not received");
            Assert.That(messages.OfType<TestEvent>().Count() == 1, "TestEvent not received");
            Assert.That(messages.OfType<TestRequest>().Count() == 1, "TestRequest not received");
            Assert.That(messages.OfType<TestResponse>().Count() == 1, "TestResponse not received");

            subscriptions.Dispose();
            ((IDisposable)serviceBus).Dispose();
            // win!
        }