protected AppendEntryBaseTest()
        {
            // Arrange  
            var mocker = new AutoMocker();

            ServerIdentifier = new ServerIdentifier();

            mocker.Use(ServerIdentifier);

            mocker.Setup<ISubject<AppendEntryMessage>>(p => p.Subscribe(It.IsAny<IObserver<AppendEntryMessage>>()))
                .Callback<IObserver<AppendEntryMessage>>((o) =>
                {
                    AppendEntryCallback = o;
                });
            AppendEntryCallback = mocker.Get<ISubject<AppendEntryMessage>>();

            AppendEntryResult = new Subject<AppendEntryResultMessage>();
            Election = new Election();
            LogReplication = new LogReplication();
            
            AppendEntry = new Rafting.AppendEntry(AppendEntryResult,
                mocker.Get<ISubject<AppendEntryMessage>>(),
                mocker.Get<IHartbeatTimer>(),
                LogReplication,
                Election,
                mocker.Get<ILoggerFactory>(),
                new RaftOptions(),
                mocker.Get<ServerIdentifier>(),
                null);
        }

        public Leader(IElection election, 
            IHartbeatTimer hartbeat,
            IObservable<AppendEntryResultMessage> reply,
            IObserver<AppendEntryMessage> append,
            IObserver<ClientResultMessage> clientReply,
            IObservable<ClientMessage> client,
            ILogReplication logReplication, 
            Nodes nodes,
            ILoggerFactory loggerFactory,
            RaftOptions options, 
            ServerIdentifier serverIdentifier)
        {
            _isDispose = false;
            _hartbeat = hartbeat;
            _append = append;
            _clientReply = clientReply;
            _client = client;
            _logReplication = logReplication;
            _nodes = nodes;
            _options = options;
            _serverIdentifier = serverIdentifier;
            _election = election;
            _logger = loggerFactory.CreateLogger(nameof(Leader) + " " + serverIdentifier);

            if (_options.UseLogging)
                _logger.LogInformation($"{nameof(ServerStateType.Leader)}");

            // Reinitialized after election
            NextIndex = new ConcurrentDictionary<ServerIdentifier, int>();
            MatchIndex = new ConcurrentDictionary<ServerIdentifier, int>();
            _hartbeat.Leader(SendHartbeat);
            _replyDispose = reply.Subscribe(EntryReplyMessageRecived);
            _clientReplyDispose = client.Subscribe(ClientMessageRecived);
        }

 public void AddObserver(IObserver observer)
 {
     if (observer != null && !m_ObserverList.Contains(observer))
     {
         m_ObserverList.Add(observer);
     }
 }

        public void RegisterObserver(IObserver observer)

        {

            observable.RegisterObserver(observer);

        }

 public void UnRegister(IObserver anObserver)
 {
     try {
     this.observerContainer.Remove(anObserver);
       } catch {
       }
 }

 /// <summary>Constructs a Rx timer stage</summary>
 /// <param name="observer">the observer</param>
 /// <param name="initialDelay">the initial delay in milli-seconds</param>
 /// <param name="period">the period in milli-seconds</param>
 public RxTimerStage(IObserver<PeriodicEvent> observer, long initialDelay, long period)
 {
     _observer = observer;
     _timer = new Timer(period);
     _timer.Elapsed += (sender, e) => OnTimedEvent(sender, e, _observer, _value);
     _timer.Enabled = true;
 }

        public void RegisterObserver(IObserver o)

        {

            observers.Add(o);

        }

 public void Detach(IObserver observer)
 {
     if (_people.Contains(observer))
        {
             _people.Add(observer);
        }
 }

        /// <summary>
        /// Initializes a new instance of the <see cref="SinkSettings" /> class.
        /// </summary>
        /// <param name="name">The event listener name.</param>
        /// <param name="sink">The sink instance.</param>
        /// <param name="eventSources">The event sources.</param>
        /// <exception cref="ConfigurationException">Validation exceptions.</exception>
        public SinkSettings(string name, IObserver<EventEntry> sink, IEnumerable<EventSourceSettings> eventSources)
            : this(name, eventSources)
        {
            Guard.ArgumentNotNull(sink, "sink");

            this.Sink = sink;
        }

 /// <summary>
 /// ��ӹ۲���
 /// </summary>
 /// <param name="observer"></param>
 public void Add(IObserver observer)
 {
     lock (this)
     {
         root.Add(observer);
     }
 }

 public void Unsubscribe(IObserver observer)
 {
     if (_observers.Contains(observer))
     {
         _observers.Remove(observer);
     }
 }

 public virtual void AddObserver( IObserver newObserver )
 {
     if ( listOfObservers.Contains( newObserver ) ) {
         return;
     }
     listOfObservers.Add( newObserver );
 }

 public static DataReaderObservableRunner RunAsync(IDbCommand command, IObserver<IDictionary<string, object>> observer)
 {
     var instance = new DataReaderObservableRunner(command, observer);
     var asyncAction = new Action(instance.Run);
     asyncAction.BeginInvoke(asyncAction.EndInvoke, null);
     return instance;
 }

        public IObservable<List<string>> ExecuteInstall(string currentAssemblyDir, IPackage bundledPackageMetadata, IObserver<int> progress = null)
        {
            progress = progress ?? new Subject<int>();

            // NB: This bit of code is a bit clever. The binaries that WiX 
            // has installed *itself* meets the qualifications for being a
            // Squirrel update directory (a RELEASES file and the corresponding 
            // NuGet packages). 
            //
            // So, in order to reuse some code and not write the same things 
            // twice we're going to "Eigenupdate" from our own directory; 
            // UpdateManager will operate in bootstrap mode and create a 
            // local directory for us. 
            //
            // Then, we create a *new* UpdateManager whose target is the normal 
            // update URL - we can then apply delta updates against the bundled 
            // NuGet package to get up to vCurrent. The reason we go through
            // this rigamarole is so that developers don't have to rebuild the 
            // installer as often (never, technically).

            var updateUsingDeltas =
                executeInstall(currentAssemblyDir, bundledPackageMetadata, progress: progress)
                        .ToObservable()
                        .ObserveOn(RxApp.DeferredScheduler)
                        .Catch<List<string>, Exception>(ex => {
                    log.WarnException("Updating using deltas has failed", ex);
                    return executeInstall(currentAssemblyDir, bundledPackageMetadata, true, progress)
                                 .ToObservable();
            });

            return updateUsingDeltas;
        }

        private static IDisposable ScheduledMethod(IObserver<int> o)
        {
            var cancel = new CancellationDisposable();
            // internally creates a new CancellationTokenSource
            NewThreadScheduler.Default.Schedule(() =>
                                                    {
                                                        int i = 0;
                                                        for (;;)
                                                        {
                                                            Thread.Sleep(200);
                                                            // here we do the long lasting background operation
                                                            if (!cancel.Token.IsCancellationRequested)
                                                                // check cancel token periodically
                                                                o.OnNext(i++);
                                                            else
                                                            {
                                                                Console.WriteLine(
                                                                    "Aborting because cancel event was signaled!");
                                                                o.OnCompleted();
                                                                return;
                                                            }
                                                        }
                                                    }
                );

            return cancel;
        }

 /// <summary>
 /// This will test if the exception has an error code that is not 0 (0 is success), if it does then it sends the observer an OnError.
 /// </summary>
 /// <param name="observer">
 /// The observer.
 /// </param>
 public void SuccessTest(IObserver<IUsbDeviceNotification> observer)
 {
     if (!Win32Error.IsSuccess)
     {
         observer.OnError(Win32Error.Exception);
     }
 }

        public void Register(IObserver o)
        {
            _observers.Add(o);

            //When new observers registers, tell the observer what we currently got
            LoadAllWindows();
        }

        public Task<long> ReadSegments(
            IObserver<RecordSegment> observer,
            Stream stream,
            CancellationToken cancellationToken)
        {
            return Task.Factory.StartNew(() => {
                var scanner = (IScanner) Activator.CreateInstance(this.SegmentsScannerType);
                var initialOffset = stream.Position;
                var lastOffset = initialOffset;

                scanner.OffsetCallback = (offset, length) => {
                    if (length > 0 && offset >= 0) {
                        var segment = new RecordSegment(initialOffset + offset, length);
                        lastOffset = initialOffset + offset + length;
                        observer.OnNext(segment);
                    }
                };

                scanner.Diagnostics = this.Diagnostics ?? TextWriter.Null;

                scanner.SetSourceWithEncoding(
                    new InternalStreamWrapper(stream, stream.Position),
                    Encoding.UTF8.CodePage);

                scanner.Parse(cancellationToken);

                return lastOffset;
            }, cancellationToken,
                TaskCreationOptions.LongRunning,
                this.TaskScheduler);
        }

 public void Attach(IObserver observerMe)
 {
     if (!_observers.Contains(observerMe))
     {
         _observers.Add(observerMe);
     }
 }

        public async Task RunAsync(IObserver<string> observer)
        {
            var urls = new[]
            {
                "http://yahoo.co.jp",
                "http://google.co.jp",
                "http://bing.co.jp",
                "http://awsedrftgyhujikol.jp/",
            };

            foreach (var url in urls)
            {
                await Observable.Timer(TimeSpan.FromSeconds(1));

                try
                {
                    var res = await GetAsStringAsync(url);
                    observer.OnNext(res);
                }
                catch(Exception ex)
                {
                    observer.OnError(ex);
                }
            }
        }