// GetViewerURLForStreams returns a log stream viewer URL to the aggregate set
// of supplied streams.
//
// If the any of the Prefix, Project, or CoordinatorHost values is not
// populated, an error will be returned.
184765
func (bs *Bootstrap) GetViewerURLForStreams(streams ...streamclient.Stream) (string, error) {
	if bs.Prefix == "" {
		return "", errors.New("no prefix is configured")
	}

	paths := make([]types.StreamPath, len(streams))
	for i, s := range streams {
		paths[i] = bs.Prefix.Join(types.StreamName(s.Properties().Name))
	}
	return bs.GetViewerURL(paths...)
}
91525
func (logger *StreamLogger) ServeLogs(url string, logf LogFormatter) {
	http.HandleFunc(url, func(w http.ResponseWriter, r *http.Request) {
		if err := acl.CheckAccessHTTP(r, acl.DEBUGGING); err != nil {
			acl.SendError(w, err)
			return
		}
		if err := r.ParseForm(); err != nil {
			http.Error(w, err.Error(), http.StatusBadRequest)
		}
		ch := logger.Subscribe("ServeLogs")
		defer logger.Unsubscribe(ch)

		// Notify client that we're set up. Helpful to distinguish low-traffic streams from connection issues.
		w.WriteHeader(http.StatusOK)
		w.(http.Flusher).Flush()

		for message := range ch {
			if err := logf(w, r.Form, message); err != nil {
				return
			}
			w.(http.Flusher).Flush()
		}
	})
	log.Infof("Streaming logs from %s at %v.", logger.Name(), url)
}
31644
func (s *Server) StreamAggregatedResources(stream ADSStream) error {
	// a channel for receiving incoming requests
	reqCh := make(chan *envoy.DiscoveryRequest)
	reqStop := int32(0)
	go func() {
		for {
			req, err := stream.Recv()
			if atomic.LoadInt32(&reqStop) != 0 {
				return
			}
			if err != nil {
				close(reqCh)
				return
			}
			reqCh <- req
		}
	}()

	err := s.process(stream, reqCh)
	if err != nil {
		s.Logger.Printf("[DEBUG] Error handling ADS stream: %s", err)
	}

	// prevents writing to a closed channel if send failed on blocked recv
	atomic.StoreInt32(&reqStop, 1)

	return err
}
268569
func StreamClientInterceptor(logger *zap.Logger, opts ...Option) grpc.StreamClientInterceptor {
	o := evaluateClientOpt(opts)
	return func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
		fields := newClientLoggerFields(ctx, method)
		startTime := time.Now()
		clientStream, err := streamer(ctx, desc, cc, method, opts...)
		logFinalClientLine(o, logger.With(fields...), startTime, err, "finished client streaming call")
		return clientStream, err
	}
}
520776
func (reader PluginConfigReaderWithError) GetStreamMap(key string, defaultValue string) (map[MessageStreamID]string, error) {
	key = reader.config.registerKey(key)
	streamMap := make(map[MessageStreamID]string)
	if defaultValue != "" {
		streamMap[WildcardStreamID] = defaultValue
	}

	if reader.HasValue(key) {
		value, err := reader.config.Settings.StringMap(key)
		if err != nil {
			return nil, err
		}

		for streamName, target := range value {
			streamMap[GetStreamID(streamName)] = target
		}
	}

	return streamMap, nil
}
591734
func NewURLBuilder(addr *types.StreamAddr) *ViewerURLBuilder {
	prefix, _ := addr.Path.Split()
	return &ViewerURLBuilder{
		Host:    addr.Host,
		Prefix:  prefix,
		Project: addr.Project,
	}
}
697593
func (c *Contributor) GetFollowingURL() string {
	if c == nil || c.FollowingURL == nil {
		return ""
	}
	return *c.FollowingURL
}
262278
func newStreamWatcher(r io.ReadCloser) watchInterface {
	sw := &streamWatcher{
		r:       r,
		decoder: json.NewDecoder(r),
		result:  make(chan Event),
	}
	go sw.receive()
	return sw
}
1158920
func (l *Client) GetStreams(stream string) ([]*Stream, error) {
	res := ApiResp{}
	data := []*Stream{}

	// make the request
	err := l.Get(parseLyticsURL(routeSchemaStreams, nil), nil, nil, &res, &data)
	if err != nil {
		return data, err
	}

	return data, nil
}
727143
func (u *User) GetFollowingURL() string {
	if u == nil || u.FollowingURL == nil {
		return ""
	}
	return *u.FollowingURL
}
1180980
func (r *Repository) GetCollaboratorsURL() string {
	if r == nil || r.CollaboratorsURL == nil {
		return ""
	}
	return *r.CollaboratorsURL
}
503322
func GCSWatcherWithURLs(apiURL, tqURL string) GCSWatcherOption {
	return &gcsWatcherURLOption{
		APIObjectChangeNotificationURL: apiURL,
		ObjectToBigQueryURL:            tqURL,
	}
}
562348
func StreamServerInterceptor(opts ...Option) grpc.StreamServerInterceptor {
	o := evaluateOptions(opts)
	return func(srv interface{}, stream grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
		if o.filterOutFunc != nil && !o.filterOutFunc(stream.Context(), info.FullMethod) {
			return handler(srv, stream)
		}
		newCtx, serverSpan := newServerSpanFromInbound(stream.Context(), o.tracer, info.FullMethod)
		wrappedStream := grpc_middleware.WrapServerStream(stream)
		wrappedStream.WrappedContext = newCtx
		err := handler(srv, wrappedStream)
		finishServerSpan(newCtx, serverSpan, err)
		return err
	}
}
1231095
func readFileOrURL(workingDir, fileName string) ([]byte, error) {
	// read from URL if it is an URL, otherwise read from local file.
	if isURL(fileName) {
		return readURL(fileName)
	}
	return readFileContents(workingDir, fileName)
}
20235
func getSourceStreamFromURL(urlStr string, encKeyDB map[string][]prefixSSEPair) (reader io.ReadCloser, err *probe.Error) {
	alias, urlStrFull, _, err := expandAlias(urlStr)
	if err != nil {
		return nil, err.Trace(urlStr)
	}
	sse := getSSE(urlStr, encKeyDB[alias])
	reader, _, err = getSourceStream(alias, urlStrFull, false, sse)
	return reader, err
}
179449
func (t *Template) GetAllAWSAppStreamUserResources() map[string]*resources.AWSAppStreamUser {
	results := map[string]*resources.AWSAppStreamUser{}
	for name, untyped := range t.Resources {
		switch resource := untyped.(type) {
		case *resources.AWSAppStreamUser:
			results[name] = resource
		}
	}
	return results
}
321326
func (b *BasicAuth) StreamInterceptor(srv interface{}, stream grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
	ctx := stream.Context()
	if err := b.doAuth(ctx); err != nil {
		return err
	}
	return handler(srv, stream)
}