// SubscribeInterceptor returns a subscriber middleware with added logging via Zap
509682
func (o Middleware) SubscribeInterceptor(opts pubsub.HandlerOptions, next pubsub.MsgHandler) pubsub.MsgHandler {
	return func(ctx context.Context, m pubsub.Msg) error {
		if o.Logger == nil {
			o.Logger = DefaultLogger()
		}

		start := time.Now()
		err := next(ctx, m)
		elapsed := time.Now().Sub(start)

		var traceID string
		span := opentracing.SpanFromContext(ctx)
		if span != nil {
			zs, ok := span.Context().(zipkintracing.SpanContext)
			if ok {
				traceID = zs.TraceID.ToHex()
			}
		}

		fields := []zapcore.Field{
			zap.String("component", "pubsub"),
			zap.String("topic", opts.Topic),
			zap.String("handler", opts.Name),
			zap.Duration("duration", elapsed),
			zap.Error(err),
		}

		if m.ID != "" {
			fields = append(fields, zap.String("id", m.ID))
		}

		if traceID != "" {
			fields = append(fields, zap.String("trace-id", traceID))
		}

		o.Logger.Debug("Processed PubSub Msg",
			fields...,
		)

		return err
	}
}
933546
func (c *Client) Subscribe(ctx context.Context, req *SubscriptionRequest) (string, error) {
	c.subscriptions.add(req)
	err := c.asynchronous.Send(ctx, req)
	if err != nil {
		// propagate send error
		return "", err
	}
	return req.SubID, nil
}
638239
func UnaryServerInterceptor(logger *zap.Logger, opts ...Option) grpc.UnaryServerInterceptor {
	o := evaluateServerOpt(opts)
	return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
		startTime := time.Now()

		newCtx := newLoggerForCall(ctx, logger, info.FullMethod, startTime)

		resp, err := handler(newCtx, req)
		if !o.shouldLog(info.FullMethod, err) {
			return resp, err
		}
		code := o.codeFunc(err)
		level := o.levelFunc(code)

		// re-extract logger from newCtx, as it may have extra fields that changed in the holder.
		ctx_zap.Extract(newCtx).Check(level, "finished unary call with code "+code.String()).Write(
			zap.Error(err),
			zap.String("grpc.code", code.String()),
			o.durationFunc(time.Since(startTime)),
		)

		return resp, err
	}
}
1307539
func (m *SubscriptionManager) subscriptionHandler() {
	defer m.wg.Done()

	for {
		select {
		// A new subscription request has been received from a client.
		case msg := <-m.newSubscriptions:
			msg.errChan <- m.handleNewSubscription(msg)

		// A request to cancel an existing subscription has been
		// received from a client.
		case msg := <-m.cancelSubscriptions:
			m.handleCancelSubscription(msg)

		// A new block notification for the tip of the chain has been
		// received from the backing NotificationSource.
		case ntfn, ok := <-m.ntfnSource.Notifications():
			if !ok {
				log.Warn("Block source is unable to deliver " +
					"new updates")
				return
			}

			m.notifySubscribers(ntfn)

		case <-m.quit:
			return
		}
	}
}
1064717
func (o Middleware) PublisherMsgInterceptor(serviceName string, next pubsub.PublishHandler) pubsub.PublishHandler {
	return func(ctx context.Context, topic string, m *pubsub.Msg) (err error) {
		defer func() {
			if r := recover(); r != nil {
				err = recoverFrom(r, "pubsub: publish error", o.RecoveryHandlerFunc)
			}
		}()
		err = next(ctx, topic, m)
		return
	}
}
635169
func NewSubscriber(
	e endpoint.Endpoint,
	dec DecodeRequestFunc,
	enc EncodeResponseFunc,
	options ...SubscriberOption,
) *Subscriber {
	s := &Subscriber{
		e:                 e,
		dec:               dec,
		enc:               enc,
		responsePublisher: DefaultResponsePublisher,
		errorEncoder:      DefaultErrorEncoder,
		errorHandler:      transport.NewLogErrorHandler(log.NewNopLogger()),
	}
	for _, option := range options {
		option(s)
	}
	return s
}
356561
func NewSubscriberWrapper() server.SubscriberWrapper {
	return func(fn server.SubscriberFunc) server.SubscriberFunc {
		return func(ctx context.Context, p server.Message) (err error) {
			t := newPublicationTracker(p, ServerProfile)
			ctx = t.start(ctx, false)

			defer func() { t.end(ctx, err) }()

			spanCtx := getTraceFromCtx(ctx)
			if spanCtx != nil {
				ctx, t.span = trace.StartSpanWithRemoteParent(
					ctx,
					fmt.Sprintf("rpc/%s/pubsub/%s", ServerProfile.Role, p.Topic()),
					*spanCtx,
				)
			} else {
				ctx, t.span = trace.StartSpan(
					ctx,
					fmt.Sprintf("rpc/%s/pubsub/%s", ServerProfile.Role, p.Topic()),
				)
			}

			err = fn(ctx, p)
			return
		}
	}
}
827399
func (m *ClientMetrics) StreamClientInterceptor() func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
	return func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
		monitor := newClientReporter(m, clientStreamType(desc), method)
		clientStream, err := streamer(ctx, desc, cc, method, opts...)
		if err != nil {
			st, _ := status.FromError(err)
			monitor.Handled(st.Code())
			return nil, err
		}
		return &monitoredClientStream{clientStream, monitor}, nil
	}
}
526041
func StreamServerInterceptor(authFunc AuthFunc) grpc.StreamServerInterceptor {
	return func(srv interface{}, stream grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
		var newCtx context.Context
		var err error
		if overrideSrv, ok := srv.(ServiceAuthFuncOverride); ok {
			newCtx, err = overrideSrv.AuthFuncOverride(stream.Context(), info.FullMethod)
		} else {
			newCtx, err = authFunc(stream.Context())
		}
		if err != nil {
			return err
		}
		wrapped := grpc_middleware.WrapServerStream(stream)
		wrapped.WrappedContext = newCtx
		return handler(srv, wrapped)
	}
}
385130
func UnaryServerInterceptor(opts ...Option) grpc.UnaryServerInterceptor {
	o := evaluateOptions(opts)
	return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
		if o.filterOutFunc != nil && !o.filterOutFunc(ctx, info.FullMethod) {
			return handler(ctx, req)
		}
		newCtx, serverSpan := newServerSpanFromInbound(ctx, o.tracer, info.FullMethod)
		resp, err := handler(newCtx, req)
		finishServerSpan(ctx, serverSpan, err)
		return resp, err
	}
}
1163553
func (o Middleware) PublisherMsgInterceptor(serviceName string, next pubsub.PublishHandler) pubsub.PublishHandler {
	return func(ctx context.Context, topic string, m *pubsub.Msg) error {
		var user string
		audit, ok := ctx.Value("x-audit-user").(string)
		if ok && audit != "" {
			user = audit
		}

		md, ok := metadata.FromIncomingContext(ctx)
		if ok {
			if len(md["x-audit-user"]) > 0 {
				user = md["x-audit-user"][0]
			}
		}

		m.Metadata["x-audit-user"] = user
		return next(ctx, topic, m)
	}
}
1315042
func NewSubscriberWrapper() server.SubscriberWrapper {
	return func(next server.SubscriberFunc) server.SubscriberFunc {
		return func(ctx context.Context, msg server.Message) (err error) {
			t := newPublicationTracker(msg, ServerProfile)
			ctx = t.StartSpanFromContext(ctx)
			defer func() {
				t.finishWithError(err, noDebugStack)
			}()

			err = next(ctx, msg)
			return
		}
	}
}
521855
func WrapSubscriber(w SubscriberWrapper) Option {
	return func(o *Options) {
		o.SubWrappers = append(o.SubWrappers, w)
	}
}
1195782
func AddStreamInterceptor(sint grpc.StreamServerInterceptor) {
	service.StreamInts = append(service.StreamInts, sint)
}
791965
func (o Middleware) SubscribeInterceptor(opts pubsub.HandlerOptions, next pubsub.MsgHandler) pubsub.MsgHandler {
	return func(ctx context.Context, m pubsub.Msg) (err error) {
		defer func() {
			if r := recover(); r != nil {
				err = recoverFrom(r, "pubsub: subscriber panic \n", o.RecoveryHandlerFunc)
			}
		}()
		err = next(ctx, m)
		return
	}
}
933340
func StreamServerInterceptor(entry *logrus.Entry, opts ...Option) grpc.StreamServerInterceptor {
	o := evaluateServerOpt(opts)
	return func(srv interface{}, stream grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
		startTime := time.Now()
		newCtx := newLoggerForCall(stream.Context(), entry, info.FullMethod, startTime)
		wrapped := grpc_middleware.WrapServerStream(stream)
		wrapped.WrappedContext = newCtx

		err := handler(srv, wrapped)

		if !o.shouldLog(info.FullMethod, err) {
			return err
		}
		code := o.codeFunc(err)
		level := o.levelFunc(code)
		durField, durVal := o.durationFunc(time.Since(startTime))
		fields := logrus.Fields{
			"grpc.code": code.String(),
			durField:    durVal,
		}
		if err != nil {
			fields[logrus.ErrorKey] = err
		}

		levelLogf(
			ctx_logrus.Extract(newCtx).WithFields(fields), // re-extract logger from newCtx, as it may have extra fields that changed in the holder.
			level,
			"finished streaming call with code "+code.String())

		return err
	}
}
93413
func (p *Provider) Subscribe(opts pubsub.HandlerOptions, h pubsub.MsgHandler) {
	logrus.Infof("Subscribing to %s successWriter/name %s", opts.Topic, opts.ServiceName+"."+opts.Name)
	logSuccess := logrus.WithField("pubsub", "kafka")
	successWriter := logSuccess.Writer()
	p.Reader = kafka.NewReader(kafka.ReaderConfig{
		Brokers:        p.Brokers,
		GroupID:        opts.ServiceName + "." + opts.Name,
		Topic:          opts.Topic,
		MaxBytes:       10e6, // 10MB
		CommitInterval: time.Second,
		MaxWait:        50 * time.Millisecond,
		Logger:         log.New(successWriter, "", log.Lshortfile),
		ErrorLogger:    log.New(ioutil.Discard, "", log.Lshortfile),
	})

	b := &backoff.Backoff{
		Min:    200 * time.Millisecond,
		Max:    600 * time.Second,
		Factor: 2,
		Jitter: true,
	}

	go func() {
		for {
			ctx := context.Background()
			m, err := p.Reader.FetchMessage(ctx)
			if err != nil {
				d := b.Duration()
				logrus.Errorf(
					"Subscription receive to topic %s failed, reconnecting in %v. Err: %v",
					opts.Topic, d, err,
				)
				time.Sleep(d)
			}

			b.Reset()

			msg := pubsub.Msg{
				ID:   string(m.Key),
				Data: m.Value,
				Ack: func() {
					p.Reader.CommitMessages(ctx, m)
				},
				Nack: func() {},
			}

			err = h(ctx, msg)
			if err != nil {
				break
			}

			if opts.AutoAck {
				msg.Ack()
			}

			logrus.Debugf("message at topic/partition/offset %v/%v/%v\n",
				m.Topic, m.Partition, m.Offset)
		}
		successWriter.Close()
	}()
}