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) {
	
	coal.Init(stream.Model)

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

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

	
	coal.OpenStream(stream.Store, stream.Model, nil, func(e coal.Event, id bson.ObjectId, m coal.Model, token []byte) {
		
		if stream.SoftDelete && e == coal.Deleted {
			return
		}

		
		if stream.SoftDelete && e == coal.Updated {
			
			softDeleteField := coal.L(stream.Model, "fire-soft-delete", true)

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

			
			if t != nil && !t.IsZero() {
				e = coal.Deleted
			}
		}

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

		
		w.manager.broadcast(evt)
	}, nil, func(err error) bool {
		
		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 , " " , 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.
Add will add a stream to the watcher .