spark/watcher.go
func (w *Watcher) Add(stream *Stream) {
	// initialize model
	coal.Init(stream.Model)

	// check existence
	if w.streams[stream.Name()] != nil {
		panic(fmt.Sprintf(`spark: stream with name "%s" already exists`, stream.Name()))
	}

	// save stream
	w.streams[stream.Name()] = stream

	// open stream
	coal.OpenStream(stream.Store, stream.Model, nil, func(e coal.Event, id bson.ObjectId, m coal.Model, token []byte) {
		// ignore real deleted events when soft delete has been enabled
		if stream.SoftDelete && e == coal.Deleted {
			return
		}

		// handle soft deleted documents
		if stream.SoftDelete && e == coal.Updated {
			// get soft delete field
			softDeleteField := coal.L(stream.Model, "fire-soft-delete", true)

			// get deleted time
			t := m.MustGet(softDeleteField).(*time.Time)

			// change type if document has been soft deleted
			if t != nil && !t.IsZero() {
				e = coal.Deleted
			}
		}

		// create event
		evt := &Event{
			Type:   e,
			ID:     id,
			Model:  m,
			Stream: stream,
		}

		// broadcast event
		w.manager.broadcast(evt)
	}, nil, func(err error) bool {
		// report error
		w.Reporter(err)

		return true
	})
}
func (w *Watcher) Add(stream *Stream) {
	// initialize model
	coal.Init(stream.Model)

	// check existence
	if w.streams[stream.Name()] != nil {
		panic(fmt.Sprintf(`spark: stream with name "%s" already exists`, stream.Name()))
	}

	// save stream
	w.streams[stream.Name()] = stream

	// open stream
	coal.OpenStream(stream.Store, stream.Model, nil, func(e coal.Event, id bson.ObjectId, m coal.Model, token []byte) {
		// ignore real deleted events when soft delete has been enabled
		if stream.SoftDelete && e == coal.Deleted {
			return
		}

		// handle soft deleted documents
		if stream.SoftDelete && e == coal.Updated {
			// get soft delete field
			softDeleteField := coal.L(stream.Model, "fire-soft-delete", true)

			// get deleted time
			t := m.MustGet(softDeleteField).(*time.Time)

			// change type if document has been soft deleted
			if t != nil && !t.IsZero() {
				e = coal.Deleted
			}
		}

		// create event
		evt := &Event{
			Type:   e,
			ID:     id,
			Model:  m,
			Stream: stream,
		}

		// broadcast event
		w.manager.broadcast(evt)
	}, nil, func(err error) bool {
		// report error
		w.Reporter(err)

		return true
	})
}
// Add will add a stream to the watcher.