Declare a consumer.
195980
def declare_consumer(self, queue, no_ack, callback, consumer_tag,
            nowait=False):
        """"""
        return self.channel.basic_consume(queue=queue,
                                          no_ack=no_ack,
                                          callback=callback,
                                          consumer_tag=consumer_tag)
620031
@SuppressWarnings("unchecked")
  public C downstreamConsumerFactory(Supplier<Consumer<String>> downstreamConsumerFactory) {
    this.downstreamConsumerFactory = requireNonNull(downstreamConsumerFactory);
    return (C) this;
  }
903180
def declare(self, exchange, type, passive=False, durable=False,
                nowait=True, arguments=None, ticket=None, cb=None):
        
        nowait = nowait and self.allow_nowait() and not cb

        args = Writer()
        args.write_short(ticket or self.default_ticket).\
            write_shortstr(exchange).\
            write_shortstr(type).\
            write_bits(passive, durable, False, False, nowait).\
            write_table(arguments or {})
        self.send_frame(MethodFrame(self.channel_id, 40, 10, args))

        if not nowait:
            self._declare_cb.append(cb)
            self.channel.add_synchronous_cb(self._recv_declare_ok)
1068397
@Override
    public void accept(E1 former, E2 latter) {
        for (BiConsumer<E1, E2> consumer : consumers) {
            consumer.accept(former, latter);
        }
    }
684422
def initialize(self, num_of_paths=None, grid=None, seed=None):
        """"""
        super(StackedConsumer, self).initialize(grid, num_of_paths, seed)
        for c in self.consumers:
            c.initialize(grid, num_of_paths, seed)
        self.state = [c.state for c in self.consumers]
1188875
def declare_default_consuming(self, queue_name, callback, passive=False,
                                  durable=False,exclusive=False, auto_delete=False,
                                  arguments=None):
        
        result = self.declare_queue(
            queue_name=queue_name,passive=passive,
            durable=durable,exclusive=exclusive,
            auto_delete=auto_delete,arguments=arguments
        )
        self.declare_basic_consuming(
            queue_name=queue_name,
            callback=callback
        )
        return result
160864
public static <T> Publisher<T> then(Publisher<T> publisher, Consumer<T> consumer) {
        return actual -> publisher.subscribe(new CompletionAwareSubscriber<T>() {
            @Override
            protected void doOnSubscribe(Subscription subscription) {
                actual.onSubscribe(subscription);
            }

            @Override
            protected void doOnNext(T message) {
                try {
                    actual.onNext(message);
                    consumer.accept(message);
                } catch (Throwable e) {
                    onError(e);
                }
            }

            @Override
            protected void doOnError(Throwable t) {
                actual.onError(t);
            }

            @Override
            protected void doOnComplete() {
                actual.onComplete();
            }
        });
    }
1101645
def create_consumer(self):
        """"""
        with self.connection_pool.acquire(block=True) as conn:
            yield self.consumer(conn)
472166
@SuppressWarnings( "unchecked" )
    public synchronized void addConsumer( MessageConsumer<? extends Serializable> consumer ) {
        consumers.add((MessageConsumer<Serializable>)consumer);
    }
290961
@Override
  public void consume(SerializableConsumer<R> consumer) {
    checkNotNull(consumer, "consumer cannot be null");

    ConsumerStreamlet<R> consumerStreamlet = new ConsumerStreamlet<>(this, consumer);
    addChild(consumerStreamlet);
  }
739158
public static <T> void consumeIfTrue(boolean bool, T target,
            Consumer<T> consumer) {
        if (bool)
            consumer.accept(target);
    }
262106
@Override
	public void registerConsumer(LocalMessageConsumer consumer)
    {
        super.registerConsumer(consumer);
        
        subscriptionsLock.writeLock().lock();
        try
		{
        	LocalTopicSubscription subscription = subscriptionMap.remove(consumer.getSubscriberId());
        	if (subscription == null)
        	{
        		// New subscription
        		storeNewSubscription(consumer);
        	}
        	else
        	{
        		// Remove old subscription
        		removeSubscription(subscription);
        		// Replace by new subscription
        		storeNewSubscription(consumer);
        	}
		}
        finally
        {
        	subscriptionsLock.writeLock().unlock();
        }
    }
813336
def declare(self, *facts):
        

        if not self.facts:
            watchers.ENGINE.warning("Declaring fact before reset()")
        return self.__declare(*facts)
1153180
@Override
    public void setConsumer(IDataSetConsumer consumer) {
        logger.debug("setConsumer(consumer) - start");

        if (this._columnSensing) {
            _consumer = new BufferedConsumer(consumer);
        } else {
            _consumer = consumer;
        }
    }
82003
def consumer(self, service, data, to_update, status):
        
        # consumer - the service which uses the data
        service_consumer = default_provider.get_service(str(service.consumer.name.name))
        kwargs = {'user': service.user}
        getattr(service_consumer, '__init__')(service.consumer.token, **kwargs)
        instance = getattr(service_consumer, 'save_data')

        # 2) for each one
        for d in data:
            d['userservice_id'] = service.consumer.id
            # the consumer will save the data and return if success or not
            status = instance(service.id, **d)

            to_update = True

        return to_update, status
728023
synchronized void add(ConsumerSessionImpl consumer)
  {
    consumer.setId(_consumerCount);
    
    if (tc.isEntryEnabled())
      SibTr.entry(tc, "add", new Long(consumer.getIdInternal()) );

    _consumers.put(new Long(_consumerCount), consumer);

    _consumerCount++;

    if (tc.isEntryEnabled())
      SibTr.exit(tc, "add");
  }
407425
protected DispatchableKey createConsumerKey(DispatchableConsumerPoint consumerPoint,
                                            SelectionCriteria criteria,
                                            SIBUuid12 connectionUuid,
                                            boolean readAhead,
                                            boolean forwardScanning,
                                            JSConsumerSet consumerSet) throws SISelectorSyntaxException, SIDiscriminatorSyntaxException, SIResourceException
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "createConsumerKey", new Object[] { consumerPoint,
                                                          criteria,
                                                          connectionUuid,
                                                          new Boolean(readAhead),
                                                          new Boolean(forwardScanning),
                                                          consumerSet});

    RemoteQPConsumerKey consKey =
      new RemoteQPConsumerKey(consumerPoint,
                            this,criteria,
                            connectionUuid,
                            readAhead,
                            forwardScanning);
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "createConsumerKey", consKey);
    return consKey;
  }