// RunStream is called once for any open channel.  Results are shared with everyone
// subscribed to the same channel.
RunStream is called once for any open channel.  Results are shared with everyone
subscribed to the same channel.
// RunStream is called once for any open channel . Results are shared with everyone // subscribed to the same channel .
func (d *CouchbaseDatasource) RunStream(ctx context.Context, req *backend.RunStreamRequest, sender *backend.StreamSender) error {
  defer func() {
    if err := recover(); err != nil {
      log.DefaultLogger.Error(fmt.Sprintf("Panic occured: %v", err))
      debug.PrintStack()
      panic(err)
    }
  }()
	log.DefaultLogger.Info("RunStream called", "request", req)
  channel := "ds/" + d.Instance.UID + "/" + req.Path
  if query_data, present := channels[channel]; !present {
    log.DefaultLogger.Error("Failed to restore stream query")
    panic(errors.New("Failed to restore query data for channel " + channel))
  } else {
    log.DefaultLogger.Info("restored query data", query_data, "for channel", channel)
    for {
      select {
      case <-ctx.Done():
        log.DefaultLogger.Info("Context done, finish streaming", "path", req.Path)
        delete(channels, channel)
        return nil
      case <-time.After(time.Second):
        query_data.Range.From = query_data.Range.To
        query_data.Range.To = time.Now()
        resp := d.query(&channel, query_data)
        err := sender.SendFrame(resp.Frames[0], data.IncludeAll)
        if err != nil {
          log.DefaultLogger.Error("Error sending frame", "error", err)
          continue
        }
      }
    }
  }
}
func ( d * CouchbaseDatasource ) RunStream ( ctx context . Context , req * backend . RunStreamRequest , sender * backend . StreamSender ) error { defer func ( ) { if err := recover ( ) ; err != nil { log . DefaultLogger . Error ( fmt . Sprintf ( " " , err ) ) debug . PrintStack ( ) panic ( err ) } } ( ) log . DefaultLogger . Info ( " " , " " , req ) channel := " " + d . Instance . UID + " " + req . Path if query_data , present := channels [ channel ] ; ! present { log . DefaultLogger . Error ( " " ) panic ( errors . New ( " " + channel ) ) } else { log . DefaultLogger . Info ( " " , query_data , " " , channel ) for { select { case <- ctx . Done ( ) : log . DefaultLogger . Info ( " " , " " , req . Path ) delete ( channels , channel ) return nil case <- time . After ( time . Second ) : query_data . Range . From = query_data . Range . To query_data . Range . To = time . Now ( ) resp := d . query ( & channel , query_data ) err := sender . SendFrame ( resp . Frames [ 0 ] , data . IncludeAll ) if err != nil { log . DefaultLogger . Error ( " " , " " , err ) continue } } } } }
RunStream is called once for any open channel.
RunStream is called once for any open channel .
// Stream data frames periodically till stream closed by Grafana.  for { // Send new data periodically.