// AddObserver implements the AddObserver method of the eventhorizon.EventBus interface.
321108
func (b *EventBus) AddObserver(m eh.EventMatcher, h eh.EventHandler) {
	sub := b.subscription(m, h, true)
	go b.handle(m, h, sub)
}
1248212
func (d *DockerMonitor) addHandler(event Event, handler EventHandler) {
	d.handlers[event] = handler
}
1089925
func NewRequestObserver(ctx RequestObserverContext) *RequestObserver {
	// Ideally we should have a logging context so we can log into the correct
	// model rather than the api server for everything.
	module := ctx.Logger.Name()
	return &RequestObserver{
		clock:      ctx.Clock,
		hub:        ctx.Hub,
		logger:     ctx.Logger,
		connLogger: loggo.GetLogger(module + ".connection"),
		pingLogger: loggo.GetLogger(module + ".ping"),
	}
}
136490
func (worker *Worker) AddServer(net, addr string) (err error) {
	// Create a new job server's client as a agent of server
	a, err := newAgent(net, addr, worker)
	if err != nil {
		return err
	}
	worker.agents = append(worker.agents, a)
	return
}
82308
func (s *Session) addEventHandlerOnce(eventHandler EventHandler) func() {
	s.handlersMu.Lock()
	defer s.handlersMu.Unlock()

	if s.onceHandlers == nil {
		s.onceHandlers = map[string][]*eventHandlerInstance{}
	}

	ehi := &eventHandlerInstance{eventHandler}
	s.onceHandlers[eventHandler.Type()] = append(s.onceHandlers[eventHandler.Type()], ehi)

	return func() {
		s.removeEventHandlerInstance(eventHandler.Type(), ehi)
	}
}
58155
func (conn *FakeConnection) AddObject(name, path string, handler FakeHandler) {
	conn.objects[name+path] = &fakeObject{handler}
}
203842
func (o *Observer) RPCObserver() rpc.Observer {
	return &rpcObserver{
		clock:   o.clock,
		metrics: o.metrics,
	}
}
283761
func (s *Logging) SetObserver(observer Observer) {
	s.mutex.Lock()
	s.observer = observer
	s.mutex.Unlock()
}
400543
func (md *RootMetadata) AddOp(o op) {
	md.data.Changes.AddOp(o)
}
1344323
func (b *EventBus) subscription(m eh.EventMatcher, h eh.EventHandler, observer bool) *pubsub.Subscription {
	b.registeredMu.Lock()
	defer b.registeredMu.Unlock()

	if m == nil {
		panic("matcher can't be nil")
	}
	if h == nil {
		panic("handler can't be nil")
	}
	if _, ok := b.registered[h.HandlerType()]; ok {
		panic(fmt.Sprintf("multiple registrations for %s", h.HandlerType()))
	}
	b.registered[h.HandlerType()] = struct{}{}

	id := string(h.HandlerType())
	if observer { // Generate unique ID for each observer.
		id = fmt.Sprintf("%s-%s", id, uuid.New())
	}

	ctx := context.Background()

	// Get or create the subscription.
	subscriptionID := b.appID + "_" + id
	sub := b.client.Subscription(subscriptionID)
	if ok, err := sub.Exists(ctx); err != nil {
		panic("could not check subscription: " + err.Error())
	} else if !ok {
		if sub, err = b.client.CreateSubscription(ctx, subscriptionID,
			pubsub.SubscriptionConfig{
				Topic:       b.topic,
				AckDeadline: 60 * time.Second,
			},
		); err != nil {
			panic("could not create subscription: " + err.Error())
		}
	}

	return sub
}
662802
func (b *Bus) AddConsumer(c Consumer) {
	if b.stopped {
		return
	}

	_, ok := b.consumers[c]
	if ok {
		// Already consuming, ignore
		return
	}

	b.mutex.Lock()
	b.consumers[c] = make(chan Event, b.ConsumerCapacity)
	b.mutex.Unlock()

	c.StartConsuming(b.consumers[c])
}
152131
func (i *Infrastructure) AddServiceServer(serviceName string, addr Address) error {
	i.Lock.RLock()
	if _, ok := i.Services[serviceName]; !ok {
		i.Lock.RUnlock()
		return ErrServiceNotExists
	}
	i.Lock.RUnlock()
	i.Lock.Lock()
	if i.Services[serviceName].Addresses == nil {
		i.Services[serviceName].Addresses = make([]Address, 0)
	}
	i.UpdateTimestamp = time.Now().UnixNano()
	i.Services[serviceName].Addresses = append(i.Services[serviceName].Addresses, addr)
	i.Lock.Unlock()
	return nil
}
1181785
func (s *Application) AddEvent(eventName string, handler interface{}) *Application {
	if h, ok := handler.(func(func(bool), ...interface{})); ok {
		events.AddEvent(eventName, h)
		return s
	}
	if h, ok := handler.(func(...interface{}) bool); ok {
		events.AddEvent(eventName, func(next func(bool), sessions ...interface{}) {
			next(h(sessions...))
		})
		return s
	}
	s.Core.Logger().Warnf(`Invalid event function: %T`, handler)
	return s
}
807025
func (collector *serverInterceptorBuilder) Add(s grpc.StreamServerInterceptor, u grpc.UnaryServerInterceptor) {
	collector.streamInterceptors = append(collector.streamInterceptors, s)
	collector.unaryInterceptors = append(collector.unaryInterceptors, u)
}
504879
func (s *ProxyLBSetting) AddServer(ip string, port int, enabled bool) {
	var record ProxyLBServer
	var isExist = false
	for i := range s.Servers {
		if s.Servers[i].IPAddress == ip && s.Servers[i].Port == port {
			isExist = true
			s.Servers[i].Enabled = enabled
		}
	}

	if !isExist {
		record = ProxyLBServer{
			IPAddress: ip,
			Port:      port,
			Enabled:   enabled,
		}
		s.Servers = append(s.Servers, record)
	}
}
189142
func (d *eventDispatcher) AddEventListener(typ string, listener EventListener) {
	d.Lock()
	defer d.Unlock()
	d.listeners[typ] = append(d.listeners[typ], listener)
}
1018516
func (s *LoadBalancerSetting) AddServer(server *LoadBalancerServer) {
	if s.Servers == nil {
		s.Servers = []*LoadBalancerServer{}
	}
	s.Servers = append(s.Servers, server)
}