// SubscribeInterceptor returns a subscriber middleware with context of an audit user
160529
func (o Middleware) SubscribeInterceptor(opts pubsub.HandlerOptions, next pubsub.MsgHandler) pubsub.MsgHandler {
	return func(ctx context.Context, m pubsub.Msg) error {
		ctx = context.WithValue(ctx, "x-audit-user", m.Metadata["x-audit-user"])
		ctx = metadata.NewIncomingContext(ctx, metadata.New(m.Metadata))
		return next(ctx, m)
	}
}
857153
func (m *ServerMetrics) UnaryServerInterceptor() func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
	return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
		monitor := newServerReporter(m, Unary, info.FullMethod)
		monitor.ReceivedMessage()
		resp, err := handler(ctx, req)
		st, _ := status.FromError(err)
		monitor.Handled(st.Code())
		if err == nil {
			monitor.SentMessage()
		}
		return resp, err
	}
}
1234006
func (s *Server) subscriptionHandler() {
	defer s.wg.Done()

	for {
		select {

		// If a client update is received, the either a new
		// subscription becomes active, or we cancel and existing one.
		case update := <-s.clientUpdates:
			clientID := update.clientID

			// In case this is a cancellation, stop the client's
			// underlying queue, and remove the client from the set
			// of active subscription clients.
			if update.cancel {
				client, ok := s.clients[update.clientID]
				if ok {
					client.updates.Stop()
					close(client.quit)
					delete(s.clients, clientID)
				}

				continue
			}

			// If this was not a cancellation, start the underlying
			// queue and add the client to our set of subscription
			// clients. It will be notified about any new updates
			// the server receives.
			update.client.updates.Start()
			s.clients[update.clientID] = update.client

		// A new update was received, forward it to all active clients.
		case upd := <-s.updates:
			for _, client := range s.clients {
				select {
				case client.updates.ChanIn() <- upd:
				case <-client.quit:
				case <-s.quit:
					return
				}
			}

		// In case the server is shutting down, stop the clients and
		// close the quit channels to notify them.
		case <-s.quit:
			for _, client := range s.clients {
				client.updates.Stop()
				close(client.quit)
			}
			return
		}
	}
}
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
	}
}
271116
func (c *Client) SubscribeTicker(ctx context.Context, symbol string) (string, error) {
	req := &SubscriptionRequest{
		SubID:   c.nonce.GetNonce(),
		Event:   EventSubscribe,
		Channel: ChanTicker,
		Symbol:  symbol,
	}
	return c.Subscribe(ctx, req)
}
179212
func (em *clientEventMonitor) Subscribe(ID string) (*Subscription, error) {
	em.Lock()
	defer em.Unlock()

	s := &Subscription{
		ID:            ID,
		cancelChannel: make(chan chan struct{}),
		eventChannel:  make(chan *dockerclient.APIEvents),
		monitorDone:   em.done,
		handlers:      make(map[string]HandlerFunc),
		monitor:       em,
		lock:          sync.RWMutex{},
	}

	em.subscriptions[ID] = append(em.subscriptions[ID], s)
	s.run()

	return s, nil
}
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)
	}
}
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
	}
}
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
	}
}
1090949
func (r User_Customer_OpenIdConnect) AddNotificationSubscriber(notificationKeyName *string) (resp bool, err error) {
	params := []interface{}{
		notificationKeyName,
	}
	err = r.Session.DoRequest("SoftLayer_User_Customer_OpenIdConnect", "addNotificationSubscriber", params, &r.Options, &resp)
	return
}
1077555
func (m *ClientMetrics) UnaryClientInterceptor() func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
	return func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
		monitor := newClientReporter(m, Unary, method)
		monitor.SentMessage()
		err := invoker(ctx, method, req, reply, cc, opts...)
		if err != nil {
			monitor.ReceivedMessage()
		}
		st, _ := status.FromError(err)
		monitor.Handled(st.Code())
		return err
	}
}
659967
func (o Middleware) PublisherMsgInterceptor(serviceName string, next pubsub.PublishHandler) pubsub.PublishHandler {
	return func(ctx context.Context, topic string, m *pubsub.Msg) error {
		var tracer = o.Tracer
		if tracer == nil {
			tracer = opentracing.GlobalTracer()
		}

		span := spanFromContext(ctx, tracer, topic)
		defer span.Finish()

		if m.Metadata == nil {
			m.Metadata = map[string]string{}
		}

		tracer.Inject(
			span.Context(),
			opentracing.TextMap,
			opentracing.TextMapCarrier(m.Metadata))

		err := next(ctx, topic, m)
		if err != nil {
			span.SetTag("error", "true")
			span.SetTag("stack", err.Error())
		}

		return err
	}
}
1339403
func unaryInterceptor(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
	var credsConfigured bool
	for _, o := range opts {
		_, ok := o.(grpc.PerRPCCredsCallOption)
		if ok {
			credsConfigured = true
			break
		}
	}
	if !credsConfigured {
		opts = append(opts, grpc.PerRPCCredentials(oauth.NewOauthAccess(&oauth2.Token{
			AccessToken: fallbackToken,
		})))
	}
	start := time.Now()
	err := invoker(ctx, method, req, reply, cc, opts...)
	end := time.Now()
	logger("RPC: %s, start time: %s, end time: %s, err: %v", method, start.Format("Basic"), end.Format(time.RFC3339), err)
	return err
}
1079095
func UnaryInterceptor(i UnaryServerInterceptor) ServerOption {
	return newFuncServerOption(func(o *serverOptions) {
		if o.unaryInt != nil {
			panic("The unary server interceptor was already set and may not be reset.")
		}
		o.unaryInt = i
	})
}
432077
func (o Middleware) PublisherMsgInterceptor(serviceName string, next pubsub.PublishHandler) pubsub.PublishHandler {
	return func(ctx context.Context, topic string, m *pubsub.Msg) error {
		if o.Logger == nil {
			o.Logger = DefaultLogger()
		}

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

		o.Logger.Debug("Published PubSub Msg",
			zap.String("component", "pubsub"),
			zap.String("topic", topic),
			zap.Duration("duration", elapsed),
			zap.Error(err),
		)

		return 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)
	}
}
999459
func (_m *Auth) StreamInterceptor(srv interface{}, stream grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
	ret := _m.Called(srv, stream, info, handler)

	var r0 error
	if rf, ok := ret.Get(0).(func(interface{}, grpc.ServerStream, *grpc.StreamServerInfo, grpc.StreamHandler) error); ok {
		r0 = rf(srv, stream, info, handler)
	} else {
		r0 = ret.Error(0)
	}

	return r0
}