Function NewService

sqlchain/observer/service.go:116–172

NewService creates new observer service and load previous subscription from the meta database.

()

// NewService creates new observer service and load previous subscription from the meta database.
func NewService() (service *Service, err error) {
	// open observer database
	dbFile := filepath.Join(conf.GConf.WorkingRoot, dbFileName)

	db, err := xs.NewSqlite(dbFile)
	if err != nil {
		return
	}

	defer func() {
		if err != nil {
			_ = db.Close()
		}
	}()

	for _, q := range initTableSQL {
		if _, err = db.Writer().Exec(q); err != nil {
			err = errors.Wrap(err, "init table failed")
			return
		}
	}

	// init service
	service = &Service{
		db:     db,
		caller: rpc.NewCallerWithPool(mux.GetSessionPoolInstance()),
	}

	// load previous subscriptions
	rows, err := db.Writer().Query(getAllSubscriptionsSQL)
	if err != nil {
		err = errors.Wrap(err, "query previous subscriptions failed")
		return
	}

	defer func() {
		_ = rows.Close()
	}()

	for rows.Next() {
		var (
			rawDBID string
			count   int32
		)

		if err = rows.Scan(&rawDBID, &count); err != nil {
			err = errors.Wrap(err, "scan subscriptions failed")
			return
		}

		dbID := proto.DatabaseID(rawDBID)

		service.subscription.Store(dbID, newSubscribeWorker(dbID, count, service))
	}

	return
}