public override void PostCommit(ICommit committed)
 {
     if (committed != null)
     {
         _scheduler.ScheduleDispatch(committed);
     }
 }

 private void ValidateCommitStamp(ICommit commit)
 {
     if (commit.CommitStamp.AddSeconds(-2) > CommitStampProcessed)
     {
         throw new Exception(string.Format("Commit {0} {1} marked with checkpoint can not be found", CommitIdProcessed, CommitStampProcessed));
     }
 }

 public override ICommit Select(ICommit committed)
 {
     bool converted = false;
     var eventMessages = committed
         .Events
         .Select(eventMessage =>
         {
             object convert = Convert(eventMessage.Body);
             if (ReferenceEquals(convert, eventMessage.Body))
             {
                 return eventMessage;
             }
             converted = true;
             return new EventMessage { Headers = eventMessage.Headers, Body = convert };
         })
         .ToList();
     if (!converted)
     {
         return committed;
     }
     return new Commit(committed.BucketId,
         committed.StreamId,
         committed.StreamRevision,
         committed.CommitId,
         committed.CommitSequence,
         committed.CommitStamp,
         committed.CheckpointToken,
         committed.Headers,
         eventMessages);
 }

 public void Dispatch(ICommit commit)
 {
     foreach (var e in commit.Events.Select(x=>x.Body))
     {
         EventPublisher.Publish(e as IEvent);
     }
 }

        public void Save(ICommit commit) {
            if (commit == null)
                throw new ArgumentNullException("commit");

            var id = Serializer.Serialize(commit.Id);
            var value = Serializer.Serialize(commit);

            using (var connection = DbConnectionFactory.CreateOpenConnection())
            using (var trans = connection.BeginTransaction(IsolationLevel.Serializable))
            using (var command = connection.CreateCommand()) {
                command.Transaction = trans;
                command.CommandText = "INSERT INTO SqlCommits (Id, Value) VALUES (@Id, @Value)";
                command.Parameters.Add(command.CreateParameter());
                command.Parameters[0].ParameterName = "Id";
                command.Parameters[0].Value = id;
                command.Parameters.Add(command.CreateParameter());
                command.Parameters[1].ParameterName = "Value";
                command.Parameters[1].Value = value;

                if (command.ExecuteNonQuery() == 0)
                    throw new InvalidOperationException("Failed to insert a row");

                trans.Commit();
            }
        }

 public static void DispatchCommit(IPublisher bus, ICommit commit)
 {
     foreach (var @event in commit.Events)
     {
         bus.Publish(@event.Body);
     }
 }

        public void Apply(ICommit commit) {
            foreach (var addObject in commit.AddedPocos) {
                Server.MetaStore.AddNew(addObject.Meta);
                Server.IndexManager.NotifyMetaChange(addObject.Meta);
            }

            foreach (var propertySet in commit.UpdatedPocos) {
                var meta = Server.MetaStore.GetWritable(propertySet.Key);
                foreach (var setProperty in propertySet) {
                    meta.Properties[setProperty.Property] = setProperty.Value;
                }

                Server.MetaStore.Update(meta);
            }

            foreach (var addToCollection in commit.CollectionAdditions) {
                var meta = Server.MetaStore.GetWritable(addToCollection.CollectionId);
                meta.Collection.Add(addToCollection.Value);

                Server.MetaStore.Update(meta);
            }

            foreach (var removeFromCollection in commit.CollectionRemovals) {
                var meta = Server.MetaStore.GetWritable(removeFromCollection.CollectionId);
                meta.Collection.Remove(removeFromCollection.Value);

                Server.MetaStore.Update(meta);
            }
        }

 public void Dispatch(ICommit commit)
 {
     foreach (var eventMessage in commit.Events)
     {
         var msg = (IMessage) eventMessage.Body;
         _bus.Publish(msg);
     }
 }

 public override void ScheduleDispatch(ICommit commit)
 {
     if (!Started)
     {
         throw new InvalidOperationException(Messages.SchedulerNotStarted);
     }
     Logger.Info(Resources.SchedulingDelivery, commit.CommitId);
     _queue.Add(commit);
 }

        public void OnNext(ICommit value)
        {
            foreach (var ev in value.Events)
            {
                var @event = ConvertEventMessageToEvent(ev);

                Send(@event);
            }
        }

 public CommitModel(ICommit commit)
 {
     this.Sequence = commit.CommitSequence;
     this.CheckPoint = commit.CheckpointToken;
     this.Date = commit.CommitStamp;
     this.CommitId = commit.CommitId;
     this.Headers = commit.Headers;
     Events = commit.Events.ToList();
 }

 public virtual void ScheduleDispatch(ICommit commit)
 {
     if (!Started)
     {
         throw new InvalidOperationException(Messages.SchedulerNotStarted);
     }
     DispatchImmediately(commit);
     MarkAsDispatched(commit);
 }

 public override void PostCommit(ICommit committed)
 {
     Logger.DebugFormat(
         "New commit {0} on tenant {1}",
         committed.CheckpointToken,
         TenantAccessor.Current.Id
     );
     Updater.Update();
 }

 public void Dispatch(ICommit commit)
 {
     foreach (var header in commit.Headers)
     {
         if (header.Key.StartsWith("UndispatchedMessage."))
         {
             _bus.Send(header.Value);
         }
     }
 }

        public ICommit Select(ICommit committed)
        {
            var eventConversionRunner = _eventConversionRunnerFactory();

            foreach (var eventMessage in committed.Events)
            {
                eventMessage.Body = eventConversionRunner.Run(eventMessage.Body);
            }

            return committed;
        }

        public override void PostCommit(ICommit committed)
        {
            if (!committed.Headers.ContainsKey("AggregateType"))
                return;

            var type = (string)committed.Headers["AggregateType"];
            if (type != DocumentTypeName)
                return;

            HandleDocumentCreation(committed);
        }

 private void MarkAsDispatched(ICommit commit)
 {
     try
     {
         Logger.Info(Resources.MarkingCommitAsDispatched, commit.CommitId);
         _persistence.MarkCommitAsDispatched(commit);
     }
     catch (ObjectDisposedException)
     {
         Logger.Warn(Resources.UnableToMarkDispatched, commit.CommitId);
     }
 }

        public void Dispatch(ICommit commit)
        {
            foreach (var @event in commit.Events.Where(x => x.Body is IDomainEvent))
            {
                var tmp = @event;

                this.Info(() => string.Format("Attempting to dispatch event: {0}", tmp));

                _bus.Publish(@event.Body);

                this.Info(() => string.Format("Successfully dispatched event: {0}", tmp));                
            }
        }

 private void DispatchImmediately(ICommit commit)
 {
     try
     {
         Logger.Info(Resources.SchedulingDispatch, commit.CommitId);
         _dispatcher.Dispatch(commit);
     }
     catch
     {
         Logger.Error(Resources.UnableToDispatch, _dispatcher.GetType(), commit.CommitId);
         throw;
     }
 }

 public void Set(ICommit commit)
 {
     if (commit == null)
     {
         CommitIdProcessed = null;
         CommitStampProcessed = null;
     }
     else
     {
         CommitIdProcessed = commit.CommitId;
         CommitStampProcessed = commit.CommitStamp;
     }
 }