// PublisherMsgInterceptor adds logging to the publisher
760610
func (o Middleware) PublisherMsgInterceptor(serviceName string, next pubsub.PublishHandler) pubsub.PublishHandler {
	return func(ctx context.Context, topic string, m *pubsub.Msg) error {
		start := time.Now()
		err := next(ctx, topic, m)
		elapsed := time.Now().Sub(start)

		fields := logrus.Fields{
			"component": "pubsub",
			"topic":     topic,
			"duration":  elapsed,
			"metadata":  m.Metadata,
		}

		if err != nil {
			de, ok := err.(errors.DropboxError)
			if ok {
				fields["err"] = de.GetMessage()
			} else {
				fields["err"] = err
			}
		}

		logrus.WithFields(fields).Debug("Published PubSub Msg")
		return err
	}
}
1188686
func (nc *Conn) PublishMsg(m *Msg) error {
	if m == nil {
		return ErrInvalidMsg
	}
	return nc.publish(m.Subject, m.Reply, m.Data)
}
683245
func IdentityMuxer() Handler {
	return func(r Publisher, err error, data interface{}) {
		if err != nil {
			r.ReplyError(err)
			return
		}
		r.Reply(data)
	}
}
3861
func PayloadStreamClientInterceptor(logger *zap.Logger, decider grpc_logging.ClientPayloadLoggingDecider) grpc.StreamClientInterceptor {
	return func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
		if !decider(ctx, method) {
			return streamer(ctx, desc, cc, method, opts...)
		}
		logEntry := logger.With(newClientLoggerFields(ctx, method)...)
		clientStream, err := streamer(ctx, desc, cc, method, opts...)
		newStream := &loggingClientStream{ClientStream: clientStream, logger: logEntry}
		return newStream, err
	}
}
1220141
func newTerminalPrompter() *terminalPrompter {
	p := new(terminalPrompter)
	// Get the original mode before calling NewLiner.
	// This is usually regular "cooked" mode where characters echo.
	normalMode, _ := liner.TerminalMode()
	// Turn on liner. It switches to raw mode.
	p.State = liner.NewLiner()
	rawMode, err := liner.TerminalMode()
	if err != nil || !liner.TerminalSupported() {
		p.supported = false
	} else {
		p.supported = true
		p.normalMode = normalMode
		p.rawMode = rawMode
		// Switch back to normal mode while we're not prompting.
		normalMode.ApplyMode()
	}
	p.SetCtrlCAborts(true)
	p.SetTabCompletionStyle(liner.TabPrints)
	p.SetMultiLineMode(true)
	return p
}
154951
func (p *Publisher) SubReader() (reader SubReader, lastMsg interface{}) {
	p.m.Lock()
	defer p.m.Unlock()

	if p.lastMsg == nil {
		p.lastMsg = newMsg(nil)
	}
	return &subscriber{p.lastMsg.next}, p.lastMsg.val
}
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
	}
}
37508
func UnaryServerInterceptor(authFunc AuthFunc) grpc.UnaryServerInterceptor {
	return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
		var newCtx context.Context
		var err error
		if overrideSrv, ok := info.Server.(ServiceAuthFuncOverride); ok {
			newCtx, err = overrideSrv.AuthFuncOverride(ctx, info.FullMethod)
		} else {
			newCtx, err = authFunc(ctx)
		}
		if err != nil {
			return nil, err
		}
		return handler(newCtx, req)
	}
}
741710
func NewTxSubmitter(config *ClientConfig) (*TxSubmitter, error) {
	err := ValidateClientConfig(config)
	if err != nil {
		return nil, err
	}

	// TODO: make mspType configurable
	mspType := "bccsp"
	peercommon.InitCrypto(config.MspDir, config.MspId, mspType)

	Signer, err := mspmgmt.GetLocalMSP().GetDefaultSigningIdentity()
	if err != nil {
		return nil, err
	}
	creator, err := Signer.Serialize()
	if err != nil {
		return nil, err
	}

	ordererClient, err := NewOrdererClient(config)
	if err != nil {
		return nil, err
	}

	deliverClient, err := NewDeliverClient(config)
	if err != nil {
		return nil, err
	}

	return &TxSubmitter{
		Config:        config,
		Signer:        Signer,
		Creator:       creator,
		OrdererClient: ordererClient,
		DeliverClient: deliverClient,
	}, nil
}
1240364
func IdentityValueMuxer(v interface{}) Handler {
	return func(r Publisher, err error, _ interface{}) {
		if err != nil {
			r.ReplyError(err)
			return
		}
		r.Reply(v)
	}
}
450259
func (o Middleware) SubscribeInterceptor(opts pubsub.HandlerOptions, next pubsub.MsgHandler) pubsub.MsgHandler {
	return func(ctx context.Context, m pubsub.Msg) error {
		var tracer = o.Tracer
		if tracer == nil {
			tracer = opentracing.GlobalTracer()
		}

		spanContext, err := tracer.Extract(
			opentracing.TextMap,
			opentracing.TextMapCarrier(m.Metadata))

		if err != nil && err != opentracing.ErrSpanContextNotFound {
			return next(ctx, m)
		}

		handlerSpan := tracer.StartSpan(
			opts.Name,
			consumerOption{clientContext: spanContext},
			pubsubTag,
		)
		defer handlerSpan.Finish()

		ctx = opentracing.ContextWithSpan(ctx, handlerSpan)
		err = next(ctx, m)

		if err != nil {
			handlerSpan.SetTag("error", "true")
			handlerSpan.SetTag("stack", err.Error())
		}

		return err
	}
}
766339
func UnaryClientInterceptor(opts ...Option) grpc.UnaryClientInterceptor {
	o := evaluateOptions(opts)
	return func(parentCtx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
		if o.filterOutFunc != nil && !o.filterOutFunc(parentCtx, method) {
			return invoker(parentCtx, method, req, reply, cc, opts...)
		}
		newCtx, clientSpan := newClientSpanFromContext(parentCtx, o.tracer, method)
		err := invoker(newCtx, method, req, reply, cc, opts...)
		finishClientSpan(clientSpan, err)
		return err
	}
}
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)
	}
}
1117600
func UnaryClientInterceptor(entry *logrus.Entry, opts ...Option) grpc.UnaryClientInterceptor {
	o := evaluateClientOpt(opts)
	return func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
		fields := newClientLoggerFields(ctx, method)
		startTime := time.Now()
		err := invoker(ctx, method, req, reply, cc, opts...)
		logFinalClientLine(o, entry.WithFields(fields), startTime, err, "finished client unary call")
		return err
	}
}
767904
func (jf openTracingService) AddGrpcServerOptions(addInterceptors func(s grpc.StreamServerInterceptor, u grpc.UnaryServerInterceptor)) {
	addInterceptors(otgrpc.OpenTracingStreamServerInterceptor(jf.Tracer), otgrpc.OpenTracingServerInterceptor(jf.Tracer))
}
84463
func PayloadStreamClientInterceptor(entry *logrus.Entry, decider grpc_logging.ClientPayloadLoggingDecider) grpc.StreamClientInterceptor {
	return func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
		if !decider(ctx, method) {
			return streamer(ctx, desc, cc, method, opts...)
		}
		logEntry := entry.WithFields(newClientLoggerFields(ctx, method))
		clientStream, err := streamer(ctx, desc, cc, method, opts...)
		newStream := &loggingClientStream{ClientStream: clientStream, entry: logEntry}
		return newStream, err
	}
}
692252
func (sp *ServerPeer) subscribeRecvMsg(subscription spMsgSubscription) {
	sp.mtxSubscribers.Lock()
	defer sp.mtxSubscribers.Unlock()
	sp.recvSubscribers[subscription] = struct{}{}
}