Method publish

driver/oracle/extractor/extractor_oracle.go:780–820

retryOperation attempts up to `count` attempts at running given function, exiting as soon as it returns with non-error. gno: only for logging

(subject string, txMsg []byte, gno int64)

// exiting as soon as it returns with non-error.
// gno: only for logging
func (e *ExtractorOracle) publish(subject string, txMsg []byte, gno int64) (err error) {
	msgLen := len(txMsg)

	data := txMsg
	lenData := len(data)

	// lenData < NatsMaxMsg: 1 msg
	// lenData = k * NatsMaxMsg + b, where k >= 1 && b >= 0: (k+1) msg
	// b could be 0. we send a zero-len msg as a sign of termination.
	nSeg := lenData/g.NatsMaxMsg + 1
	e.logger.Debug("publish. msg", "subject", subject, "gno", gno, "nSeg", nSeg, "spanLen", lenData, "msgLen", msgLen)
	bak := make([]byte, 4)
	if nSeg > 1 {
		// ensure there are 4 bytes to save iSeg
		data = append(data, 0, 0, 0, 0)
	}
	for iSeg := 0; iSeg < nSeg; iSeg++ {
		var part []byte
		if nSeg == 1 { // not big msg
			part = data
		} else {
			begin := iSeg * g.NatsMaxMsg
			end := mathutil.Min(lenData, (iSeg+1)*g.NatsMaxMsg)
			// use extra 4 bytes to save iSeg
			if iSeg > 0 {
				copy(data[begin:begin+4], bak)
			}
			copy(bak, data[end:end+4])
			part = data[begin : end+4]
			binary.LittleEndian.PutUint32(data[end:], uint32(iSeg))
		}

		e.logger.Debug("publish", "subject", subject, "gno", gno, "partLen", len(part), "iSeg", iSeg)
		_, err := e.natsConn.Request(subject, part, 24*time.Hour)
		if err != nil {
			e.logger.Error("unexpected error on publish", "err", err)
			return err
		}
	}
	return nil
}

func (e *ExtractorOracle) onError(state int, err error) {
	e.logger.Error("onError", "err", err, "hasShutdown", e.shutdown)