Method recvProtoStream

erpc/query_pipe_through.go:124–166

(recv func() (proto.Message, error), onPage func(proto.Message) error, method string, upstreamId string)

}

func (qe *EvmQueryExecutor) recvProtoStream(recv func() (proto.Message, error), onPage func(proto.Message) error, method string, upstreamId string) error {
	type cursorPage interface {
		proto.Message
		GetCursorBlock() *evm.CursorBlock
	}

	var lastCursor *evm.CursorBlock
	pageEmitted := false
	pageCount := 0

	for {
		page, err := recv()
		if err == io.EOF {
			qe.logger.Debug().Str("upstreamId", upstreamId).Str("method", method).Int("pagesReceived", pageCount).Msgf("upstream query stream completed (EOF)")
			return nil
		}
		if err != nil {
			qe.logger.Debug().Err(err).Str("upstreamId", upstreamId).Str("method", method).Int("pagesReceived", pageCount).Bool("pageEmitted", pageEmitted).Msgf("upstream query stream error")
			return &StreamError{Err: err, LastCursor: lastCursor, PageEmitted: pageEmitted}
		}

		pageCount++
		cursorBlock := (*evm.CursorBlock)(nil)
		if cursorAware, ok := any(page).(cursorPage); ok {
			cursorBlock = cursorAware.GetCursorBlock()
			if cursorBlock != nil {
				lastCursor = cursorBlock
			}
		}

		qe.logger.Trace().Str("upstreamId", upstreamId).Str("method", method).Int("page", pageCount).Interface("cursor", cursorBlock).Msgf("received page from upstream query stream")

		if err := onPage(page); err != nil {
			return &StreamError{Err: err, LastCursor: lastCursor, PageEmitted: true}
		}
		pageEmitted = true

		if cursorBlock == nil {
			qe.logger.Debug().Str("upstreamId", upstreamId).Str("method", method).Int("pagesReceived", pageCount).Msgf("upstream query stream completed (no cursor)")
			return nil
		}
	}
}