Method Shutdown

driver/mysql/extractor.go:1561–1615

Shutdown is used to tear down the extractor

()

}
// Shutdown is used to tear down the extractor
func (e *Extractor) Shutdown() error {
	e.logger.Debug("extractor shutdown")
	e.shutdownLock.Lock()
	defer e.shutdownLock.Unlock()

	if e.shutdown {
		return nil
	}
	e.logger.Info("extractor shutdown")

	e.shutdown = true
	close(e.shutdownCh)

	if e.RevApplier != nil {
		err := e.RevApplier.Shutdown()
		if err != nil {
			e.logger.Info("error RevApplier.Shutdown", "err", err)
		}
	}

	if e.binlogReader != nil && atomic.LoadInt32(&e.binlogReader.BigTxCount) >= 1 {
		g.SubBigTxJob()
	}

	if e.natsConn != nil {
		e.natsConn.Close()
	}

	for _, d := range e.dumpers {
		d.Close()
	}

	if err := sql.CloseDB(e.singletonDB); err != nil {
		e.logger.Error("Shutdown error close singletonDB.", "err", err)
	}

	if e.inspector != nil {
		e.inspector.Close()
	}

	if e.binlogReader != nil {
		if err := e.binlogReader.Close(); err != nil {
			e.logger.Error("Shutdown error close binlogReader.", "err", err)
		}
	}

	e.wg.Wait()

	if err := sql.CloseDB(e.db); err != nil {
		e.logger.Error("Shutdown error close e.db.", "err", err)
	}

	e.logger.Info("Shutting down")
	return nil
}

func (e *Extractor) sendFullComplete() (err error) {
	dumpMsg, err := common.Encode(&common.DumpStatResult{