You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: manager.go
    path of file: ./repos/grafana/pkg/services/live/runstream
    
    the code of the file until where you have to start completion: package runstream

import (
	"context"
	"errors"
	"fmt"
	"math"
	"sync"
	"time"

	"github.com/grafana/grafana-plugin-sdk-go/backend"

	"github.com/grafana/grafana/pkg/infra/log"
	"github.com/grafan
    
func (s *Manager) watchStream(ctx context.Context, cancelFn func(), sr streamRequest) {
	numNoSubscribersChecks := 0
	presenceTicker := time.NewTicker(s.checkInterval)
	defer presenceTicker.Stop()
	datasourceTicker := time.NewTicker(s.datasourceCheckInterval)
	defer datasourceTicker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-datasourceTicker.C:
			if sr.PluginContext.DataSourceInstanceSettings != nil {
				dsUID := sr.PluginContext.DataSourceInstanceSettings.UID
				pCtx, err := s.pluginContextGetter.GetPluginContext(ctx, sr.user, sr.PluginContext.PluginID, dsUID, false)
				if err != nil {
					if errors.Is(err, plugins.ErrPluginNotRegistered) {
						logger.Debug("Datasource not found, stop stream", "channel", sr.Channel, "path", sr.Path)
						return
					}
					logger.Error("Error getting datasource context", "channel", sr.Channel, "path", sr.Path, "error", err)
					continue
				}
				if pCtx.DataSourceInstanceSettings.Updated != sr.PluginContext.DataSourceInstanceSettings.Updated {
					logger.Debug("Datasource changed, re-establish stream", "channel", sr.Channel, "path", sr.Path)
					err := s.HandleDatasourceUpdate(pCtx.OrgID, dsUID)
					if err != nil {
						logger.Error("Error re-establishing stream", "channel", sr.Channel, "path", sr.Path, "error", err)
						continue
					}
					return
				}
			}
		case <-presenceTicker.C:
			numSubscribers, err := s.presenceGetter.GetNumLocalSubscribers(sr.Channel)
			if err != nil {
				logger.Error("Error checking num subscribers", "channel", sr.Channel, "path", sr.Path, "error", err)
				continue
			}
			if numSubscribers > 0 {
				// reset counter since channel has active subscribers.
				numNoSubscribersChecks = 0
				continue
			}
			numNoSubscribersChecks++
			if numNoSubscribersChecks >= s.maxChecks {
				logger.Debug("Stop stream since no active subscribers", "channel", sr.Channel, "path", sr.Path)
				s.stopStream(sr, cancelFn)
				return
			}
		}
	}
}