// EnableClientStreamReceiveTimeHistogram turns on recording of
// single message send time of streaming RPCs.
// This function acts on the DefaultClientMetrics variable and the
// default Prometheus metrics registry.
978718
func EnableClientStreamSendTimeHistogram(opts ...HistogramOption) {
	DefaultClientMetrics.EnableClientStreamSendTimeHistogram(opts...)
	prom.Register(DefaultClientMetrics.clientStreamSendHistogram)
}
1316448
func pushMetric(addr string, interval time.Duration) {
	if interval == zeroDuration || len(addr) == 0 {
		log.Info("disable Prometheus push client")
		return
	}
	log.Info("start prometheus push client", zap.String("server addr", addr), zap.String("interval", interval.String()))
	go prometheusPushClient(addr, interval)
}
836554
func (m *ServerMetrics) StreamServerInterceptor() func(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
	return func(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
		monitor := newServerReporter(m, streamRPCType(info), info.FullMethod)
		err := handler(srv, &monitoredServerStream{ss, monitor})
		st, _ := status.FromError(err)
		monitor.Handled(st.Code())
		return err
	}
}
250162
func StreamServerInterceptor() grpc.StreamServerInterceptor {
	return func(srv interface{}, stream grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
		ctx := stream.Context()
		wrapped := grpc_middleware.WrapServerStream(stream)
		if acl.SkipAuth() {
			return handler(srv, wrapped)
		}
		//
		peer, ok := peer.FromContext(ctx)
		if !ok {
			return status.Error(codes.PermissionDenied, "")
		}
		// get tlsInfo from peer
		tlsInfo, ok := peer.AuthInfo.(credentials.TLSInfo)
		if !ok || len(tlsInfo.State.VerifiedChains) == 0 {
			return status.Error(codes.PermissionDenied, "")
		}
		ctx = client.NewContext(ctx, tlsInfo.State.VerifiedChains[0][0].SerialNumber)
		// the only Unauthenticated method is Auth.Auth
		// to access other methods JWT should have been set to context
		if !acl.CanAccessMethod(ctx, info.FullMethod) {
			//log.Infof("User %+v does not have permission to execute %s ", user.FromContext(ctx),info.FullMethod)
			return status.Errorf(codes.PermissionDenied, "User does not have permission to execute %s ", info.FullMethod)
		}
		wrapped.WrappedContext = ctx
		return handler(srv, wrapped)
	}
}
1195782
func AddStreamInterceptor(sint grpc.StreamServerInterceptor) {
	service.StreamInts = append(service.StreamInts, sint)
}
826165
func (c *criClient) EnableEventListener() (chan<- *EventMessage, error) {
	if c == nil {
		log.Debug("Not enabling CRI event listener because CRI client is nil")
		return nil, nil
	}
	log.Info("Enabling CRI event listener")

	ws := newWatcherState()

	eventsCh := make(chan *EventMessage, 100)
	go func(state *watcherState, eventsCh <-chan *EventMessage) {
		for event := range eventsCh {
			ws.enqueueByContainerID(event.WorkloadID, event)
		}
	}(ws, eventsCh)
	return eventsCh, nil
}
943985
func NewDefaultCciForStreamWithSrcLen(sourceLength uint, priceStream gotrade.DOHLCVStreamSubscriber) (indicator *Cci, err error) {
	ind, err := NewDefaultCciWithSrcLen(sourceLength)
	priceStream.AddTickSubscription(ind)
	return ind, err
}
393845
func (o *rpcObserver) ServerReply(req rpc.Request, hdr *rpc.Header, body interface{}) {
	labels := prometheus.Labels{
		MetricLabelFacade:    req.Type,
		MetricLabelVersion:   strconv.Itoa(req.Version),
		MetricLabelMethod:    req.Action,
		MetricLabelErrorCode: hdr.ErrorCode,
	}
	duration := o.clock.Now().Sub(o.requestStart)
	o.metrics.apiRequestDuration.With(labels).Observe(duration.Seconds())

	// The following is obsolete and should be removed for 2.6 release
	o.metrics.deprecatedAPIRequestDuration.With(labels).Observe(duration.Seconds())
	o.metrics.deprecatedAPIRequestsTotal.With(labels).Inc()
}
1211010
func (db *DB) LastPingReceived(id ID, ip net.IP) time.Time {
	return time.Unix(db.fetchInt64(nodeItemKey(id, ip, dbNodePing)), 0)
}
1229977
func (c *Client) RPC(method string, args interface{}, resp interface{}) error {
	// Get a stream
	stream, err := c.Open()
	if err != nil {
		return fmt.Errorf("failed to open stream: %v", err)
	}
	defer stream.Close()
	stream.SetDeadline(time.Now().Add(rpcTimeout))

	// Create the RPC client
	cc := msgpackrpc.NewCodec(true, true, stream)
	return msgpackrpc.CallWithCodec(cc, method, args, resp)
}
678355
func (m *RTMetrics) LatencyHistogram() (*HDRHistogram, error) {
	m.histogramLock.Lock()
	defer m.histogramLock.Unlock()
	return m.histogram.Merged()
}
424989
func wrapStreamRegistrationCallback(reg streamRegistrationCallback) streamRegistrationCallback {
	if reg == nil {
		return reg
	}
	return func(desc *logpb.LogStreamDescriptor) bundler.StreamChunkCallback {
		cb := reg(desc)
		switch desc.StreamType {
		case logpb.StreamType_TEXT:
			return buffered_callback.GetWrappedTextCallback(cb)
		case logpb.StreamType_DATAGRAM:
			return buffered_callback.GetWrappedDatagramCallback(cb)
		default:
			return cb
		}
	}
}
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
	}
}
1104697
func (h *MessageReceivedSizeStatsHandler) HandleRPC(ctx context.Context, stat stats.RPCStats) {
	lab, _ := ctx.Value(tagRPCKey).(prometheus.Labels)

	if pay, ok := stat.(*stats.InPayload); ok {
		switch {
		case stat.IsClient() && h.subsystem == Client:
			h.vec.With(lab).Observe(float64(pay.Length))
		case !stat.IsClient() && h.subsystem == Server:
			h.vec.With(lab).Observe(float64(pay.Length))
		}
	}
}
1276196
func httpStreamReceived(streams chan httpstream.Stream) func(httpstream.Stream, <-chan struct{}) error {
	return func(stream httpstream.Stream, replySent <-chan struct{}) error {
		// make sure it has a valid port header
		portString := stream.Headers().Get(api.PortHeader)
		if len(portString) == 0 {
			return fmt.Errorf("%q header is required", api.PortHeader)
		}
		port, err := strconv.ParseUint(portString, 10, 16)
		if err != nil {
			return fmt.Errorf("unable to parse %q as a port: %v", portString, err)
		}
		if port < 1 {
			return fmt.Errorf("port %q must be > 0", portString)
		}

		// make sure it has a valid stream type header
		streamType := stream.Headers().Get(api.StreamType)
		if len(streamType) == 0 {
			return fmt.Errorf("%q header is required", api.StreamType)
		}
		if streamType != api.StreamTypeError && streamType != api.StreamTypeData {
			return fmt.Errorf("invalid stream type %q", streamType)
		}

		streams <- stream
		return nil
	}
}
54199
func streamEventCallback(st C.virStreamPtr, events int, callbackID int) {
	callbackFunc := getCallbackId(callbackID)

	callback, ok := callbackFunc.(StreamEventCallback)
	if !ok {
		panic("Incorrect stream event func callback")
	}

	callback(&Stream{ptr: st}, StreamEventType(events))
}
484547
func RegisterStreamServiceHandler(ctx context.Context, mux *runtime.ServeMux, conn *grpc.ClientConn) error {
	return RegisterStreamServiceHandlerClient(ctx, mux, NewStreamServiceClient(conn))
}