Subscribe a handler to a channel.

@param string $channel
@param callable $handler

@throws \Exception
341039
public function subscribe($channel, callable $handler)
    {
        $this->consumer->subscribe([$channel]);

        $isSubscriptionLoopActive = true;

        while ($isSubscriptionLoopActive) {
            $message = $this->consumer->consume(120 * 1000);

            if ($message === null) {
                continue;
            }

            switch ($message->err) {
                case RD_KAFKA_RESP_ERR_NO_ERROR:
                    $payload = Utils::unserializeMessagePayload($message->payload);

                    if ($payload === 'unsubscribe') {
                        $isSubscriptionLoopActive = false;
                    } else {
                        call_user_func($handler, $payload);
                    }

                    $this->consumer->commitAsync($message);

                    break;
                case RD_KAFKA_RESP_ERR__PARTITION_EOF:
                case RD_KAFKA_RESP_ERR__TIMED_OUT:
                    break;
                default:
                    throw new \Exception($message->errstr(), $message->err);
            }
        }
    }
128089
@Override
  public void channelActive(final ChannelHandlerContext ctx) throws Exception {
    this.channelGroup.add(ctx.channel());
    this.listener.channelActive(ctx);
    super.channelActive(ctx);
  }
240915
private function load_handler_as_method($handler) {
	list($class, $method) = explode('->', $handler);
	
	$object = new $class;
	$object->$method($this->url, $this->method, $this->arguments);
}
557529
public function add(string $signal, callable $handler): self
    {
        $this->handlers[$signal] = $handler;

        return $this;
    }
264920
public void addSubscriber(Handler handler) {
        String handlerId = CollectorManagerUtils.getHandlerId(handler);

        //temporary exception for audit Source  //must change later
        if (source.getSourceName().trim().equals("audit") && subscribers.isEmpty()) {
            //First subscriber, assign a buffer.
            bufferMgr = new BufferManagerImpl(BUFFER_SIZE, sourceId, false);
            //Inform the source that a buffer is now available
            //and it can start sending events to this buffer.
            source.setBufferManager(this.bufferMgr);
        }
        subscribers.add(handlerId);
        /*
         * Inform the handler that this source/buffer/conduit is now available:
         * Synchronous Handler - Add handler as synchronous handler to the Buffer/Conduit's sync handler set
         * Asynchronous Handler - Add handler as asynchronous handler into the Buffer/Conduit
         *
         */
        if (handler instanceof SynchronousHandler) {
            bufferMgr.addSyncHandler((SynchronousHandler) handler);
        } else {
            bufferMgr.addHandler(handlerId);
        }

    }
393286
public function handle( $handler = '' ) {
		if ( ! empty( $handler ) ) {
			$this->attribute( 'handler', $handler );
		}

		$this->router->addRoute( $this->router->route( $this->getAttributes() ) );
	}
631325
public function replace($handler, $bindTo = true)
    {
        if ($bindTo && $handler instanceof Closure) {
            $handler = $handler->bindTo($this, $this);
        }
        $this->handler = $handler;
    }
784689
public function withHandler($handler): self
    {
        return tap($this, function ($request) use ($handler) {
            return $this->handler = $handler;
        });
    }
232611
public function register($name, $handler)
    {
        Assert::string($name, 'The handler name must be a string. Got: %s');
        Assert::notEmpty($name, 'The handler name must not be empty.');

        if (!is_object($handler)) {
            Assert::isCallable($handler, 'The handler must be a callable or an object. Got: %s');
        }

        $this->handlers[$name] = $handler;

        if (!$this->selectedHandler) {
            $this->selectedHandler = $name;
        }
    }
1339336
public function setExceptionHandler(callable $handler = null)
    {
        $prev = $this->exceptionHandler;
        $this->exceptionHandler = $handler;

        return $prev;
    }
517154
public function handler($handler, $construct = null)
    {
        $this->handler   = $handler;
        $this->construct = $construct;
        return $this;
    }
763709
public final boolean register(final SelectableChannel channel, final EventsHandler handler)
    {
        return register(channel, handler, IN | OUT | ERR);
    }
494526
public function subscribe(Dispatcher $events)
    {
        foreach ($this->listeners as $name => $handler)
        {
            $events->listen($name, $this->className.'@'.$handler);
        }
    }
1305479
private function handle($handler, $parameters)
    {
        if ($handler instanceof \Closure) {
            $this->app->execute($handler, $parameters);
        } else {
            list($controller, $action) = explode('@', $handler);

            $class = $this->app->construct($controller);
            $this->app->execute([$class, $action], $parameters);
        }
    }
557260
public function registerHandler($contentType, ContentHandlerInterface $handler)
    {
        if (!is_string($contentType)) {
            throw new Exception('Invalid content handler specified.');
        }

        $this->handlers[$contentType] = $handler;
    }
313888
public function addRecoverableErrorHandler($handler) {
        if ( ! $handler instanceof INativeErrorHandler && ! is_callable($handler)) {
            throw new InvalidHandlerType(
                s('Recoverable error handler must be a callable or an instance of %s.',
                    INativeErrorHandler::class)
            );
        }

        if (is_callable($handler)) {
            $handler = $this->createNativeErrorHandler($handler);
        }

        $this->recoverableErrorHandlers[] = $handler;
    }
992765
public void subscribe(final Class<? extends T> clazz, final EventHandler<? extends T> handler) {
    lock.writeLock().lock();
    try {
      List<EventHandler<? extends T>> list = clazzToListOfHandlersMap.get(clazz);
      if (list == null) {
        list = new LinkedList<EventHandler<? extends T>>();
        clazzToListOfHandlersMap.put(clazz, list);
      }
      list.add(handler);
    } finally {
      lock.writeLock().unlock();
    }
  }