Method subscribe

sqlchain/observer/service.go:174–200

(dbID proto.DatabaseID, resetSubscribePosition string)

}

func (s *Service) subscribe(dbID proto.DatabaseID, resetSubscribePosition string) (err error) {
	if atomic.LoadInt32(&s.stopped) == 1 {
		return ErrStopped
	}

	if resetSubscribePosition != "" {
		var fromPos int32

		switch resetSubscribePosition {
		case "newest":
			fromPos = types.ReplicateFromNewest
		case "oldest":
			fromPos = types.ReplicateFromBeginning
		default:
			fromPos = types.ReplicateFromNewest
		}

		unpackWorker(s.subscription.LoadOrStore(dbID,
			newSubscribeWorker(dbID, fromPos, s))).reset(fromPos)
	} else {
		// not resetting
		unpackWorker(s.subscription.LoadOrStore(dbID,
			newSubscribeWorker(dbID, types.ReplicateFromNewest, s))).start()
	}

	return
}

func unpackWorker(actual interface{}, _ ...interface{}) (worker *subscribeWorker) {
	if actual == nil {