~gabe/mortar unlisted

f52cc695956bed9f006ac323fa2646e29383a800 — Gabe Fierro 1 year, 6 months ago f5cdc47 v1.1.0-alpha3
influxdb handles cancelled context now
1 files changed, 10 insertions(+), 2 deletions(-)

M stages/influxdb.go
M stages/influxdb.go => stages/influxdb.go +10 -2
@@ 206,7 206,11 @@ func (stage *InfluxDBTimeseriesQueryStage) processQuery(req *Request) error {
					if pcount == TS_BATCH_SIZE {
						tsresp.DataFrame = dataFrame.Name
						tsresp.Identifier = uuStr
						req.fetch_responses <- tsresp
						select {
						case req.fetch_responses <- tsresp:
						case <-req.Done():
							continue
						}
						//stage.output <- ctx
						tsresp = &mortarpb.FetchResponse{}
						pcount = 0


@@ 217,7 221,11 @@ func (stage *InfluxDBTimeseriesQueryStage) processQuery(req *Request) error {
			if len(tsresp.Times) > 0 {
				tsresp.DataFrame = dataFrame.Name
				tsresp.Identifier = uuStr
				req.fetch_responses <- tsresp
				select {
				case req.fetch_responses <- tsresp:
				case <-req.Done():
					continue
				}
				//stage.output <- ctx
			}