-- pump messages on a producer
556225
function pump(p) {
  function push() {
    if (p.sendToQueue(p.qName, message)) {
      ++totalPushed;
      setTimeout(push, pushback);
    }
  }
  push();
}
366006
def create_producer(self):
        """"""
        with self.connection_pool.acquire(block=True) as conn:
            yield self.producer(conn)
350230
function (emits, message, daemonPid) {
    var monitor = new Monitor(message, options, daemonPid, shutdown, child, function () {

      // emit events before starting child
      Object.keys(emits).forEach(function (name) {
        // skip the process event until the child spawns
        if (name === 'process') return;

        monitor.emit.apply(monitor, [name].concat( emits[name] ));
      });

      // spawn child when monitor is ready
      child.spawn();

      // start pumping streams to monitor
      child.pump(monitor);
    });

    // once the child has started emit the process event
    child.once('start', function () {
      monitor.pid.process = child.pid;
      monitor.emit.apply(monitor, ['process'].concat( emits.process ));
    });

    // relay stop and restart to process event on monitor
    child.on('stop', function () {
      monitor.pid.process = null;

      // Do not emit if the deamon is in restart mode
      if (daemonInRestart) return;
      monitor.emit('process', 'stop');
    });
    child.on('restart', function () {
      monitor.pid.process = child.pid;
      monitor.emit('process', 'restart');
    });
  }
713952
function doProduce(v, k) {
    try {
      var r = self._oldProduce(topic, partition,
        v, k,
        timestamp, opaque);

      self._hl.deliveryEmitter.once(opaque.__message_id, function(err, offset) {
        self._hl.pollingRef.decrement();
        setImmediate(function() {
          // Offset must be greater than or equal to 0 otherwise it is a null offset
          // Possibly because we have acks off
          callback(err, offset >= 0 ? offset : null);
        });
      });
      
      return r;
    } catch (e) {
      callback(e);
    }
  }
1219092
def emit(self, record):
        
        try:
            message = self.format(record)

            if not self.is_colorized:
                self.stream.write(message)
            else:
                style = self._get_style_function_for_level(record.levelno)
                self.stream.write(style(message))

            self.stream.write(getattr(self, 'terminator', '\n'))
            self.flush()

        except (KeyboardInterrupt, SystemExit):
            raise
        except:
            self.handleError(record)
378668
def publish(self, msg):
        '''
        
        '''
        if not self.streams:
            return

        pack = salt.transport.frame.frame_msg_ipc(msg, raw_body=True)

        for stream in self.streams:
            self.io_loop.spawn_callback(self._write, stream, pack)
201183
private void runMessage(MessageAmp msg)
  {
    try (OutboxAmp outbox = OutboxAmpFactory.newFactory().get()) {
      outbox.inbox(msg.inboxTarget());
      outbox.message(msg);
      
      //RampActor systemActor = null;
      StubAmp systemActor = _stub;
      
      msg.invoke(InboxExecutorBase.this, systemActor);
      
      while (! outbox.flushAndExecuteLast()) {
      }
    }
  }
1335963
async def send(self, payload='', action_type='', channel=None, **kwds):
        
        # use a custom channel if one was provided
        channel = channel or self.producer_channel

        # serialize the action type for the
        message = serialize_action(action_type=action_type, payload=payload, **kwds)
        # send the message
        return await self._producer.send(channel, message.encode())
740420
def on_message(self, message):
        """"""
        json_rpc = json.loads(message)
        logging.log(logging.DEBUG, json_rpc)

        if self.pool is None:
            self.pool = multiprocessing.Pool(processes=args.workers)

        # Spawn a process to protect the server against segfaults
        async = self.pool.apply_async(_worker_process, [json_rpc])
        try:
            result = async.get(timeout=args.timeout)
            error = 0
        except multiprocessing.TimeoutError:
            result = ("File format conversion timed out! This is due "
                      "either to a large input file or a segmentation "
                      "fault in the underlying open babel library.")
            error = 1
            self.pool.terminate()
            self.pool = multiprocessing.Pool(processes=args.workers)
        except Exception:
            result = traceback.format_exc()
            error = 1
        logging.log(logging.DEBUG, result)

        self.write_message(json.dumps({'result': result, 'error': error,
                                       'id': json_rpc['id']},
                                      separators=(',', ':')))
1200519
@Override
    public void sendInternal(NotificationType type, String... messages) throws NotificationException {
        if (producer == null) {
            createProducer();
        }
        sendInternalToProducer(producer, type, messages);
    }
244748
def display_message(self, clear, beep, timeout, line1, line2):
        """"""
        self._elk.send(
            dm_encode(self._index, clear, beep, timeout, line1, line2)
        )
881716
private void sendPingFrame(ChannelHandlerContext ctx, SendPingCommand msg,
      ChannelPromise promise) {
    // Don't check lifecycleManager.getShutdownStatus() since we want to allow pings after shutdown
    // but before termination. After termination, messages will no longer arrive because the
    // pipeline clears all handlers on channel close.

    PingCallback callback = msg.callback();
    Executor executor = msg.executor();
    // we only allow one outstanding ping at a time, so just add the callback to
    // any outstanding operation
    if (ping != null) {
      promise.setSuccess();
      ping.addCallback(callback, executor);
      return;
    }

    // Use a new promise to prevent calling the callback twice on write failure: here and in
    // NettyClientTransport.ping(). It may appear strange, but it will behave the same as if
    // ping != null above.
    promise.setSuccess();
    promise = ctx().newPromise();
    // set outstanding operation
    long data = USER_PING_PAYLOAD;
    Stopwatch stopwatch = stopwatchFactory.get();
    stopwatch.start();
    ping = new Http2Ping(data, stopwatch);
    ping.addCallback(callback, executor);
    // and then write the ping
    encoder().writePing(ctx, false, USER_PING_PAYLOAD, promise);
    ctx.flush();
    final Http2Ping finalPing = ping;
    promise.addListener(new ChannelFutureListener() {
      @Override
      public void operationComplete(ChannelFuture future) throws Exception {
        if (future.isSuccess()) {
          transportTracer.reportKeepAliveSent();
        } else {
          Throwable cause = future.cause();
          if (cause instanceof ClosedChannelException) {
            cause = lifecycleManager.getShutdownThrowable();
            if (cause == null) {
              cause = Status.UNKNOWN.withDescription("Ping failed but for unknown reason.")
                  .withCause(future.cause()).asException();
            }
          }
          finalPing.failed(cause);
          if (ping == finalPing) {
            ping = null;
          }
        }
      }
    });
  }
291070
def _pump(self):
        '''
        
        '''
        while (not self._busy) and len(self._queue):
            cmd = self._queue.pop(0)
            self._name = cmd[2]
            try:
                cmd[0](*cmd[1])
            except Exception as e:
                self.notify('error', exception=e)
                if self._debug:
                    traceback.print_exc()
726524
public void receiveAndProcess() {
        if (this.shouldContinue()) {
            this.receiver.receive(this.onReceiveHandler.getMaxEventCount())
                    .handleAsync(this.processAndReschedule, this.executor);
        } else {
            if (TRACE_LOGGER.isInfoEnabled()) {
                TRACE_LOGGER.info(String.format(Locale.US, "Stopping receive pump for eventHub (%s), consumerGroup (%s), partition (%s) as %s",
                        this.eventHubName, this.consumerGroupName, this.receiver.getPartitionId(),
                        this.stopPumpRaised.get() ? "per the request." : "pump ran into errors."));
            }

            this.stopPump.complete(null);
        }
    }
1313649
def _SendTerminationMessage(self, status=None):
    """"""
    if not self.runner_args.request_state.session_id:
      # No parent flow, nothing to do here.
      return

    if status is None:
      status = rdf_flows.GrrStatus()

    client_resources = self.context.client_resources
    user_cpu = client_resources.cpu_usage.user_cpu_time
    sys_cpu = client_resources.cpu_usage.system_cpu_time
    status.cpu_time_used.user_cpu_time = user_cpu
    status.cpu_time_used.system_cpu_time = sys_cpu
    status.network_bytes_sent = self.context.network_bytes_sent
    status.child_session_id = self.session_id

    request_state = self.runner_args.request_state
    request_state.response_count += 1

    # Make a response message
    msg = rdf_flows.GrrMessage(
        session_id=request_state.session_id,
        request_id=request_state.id,
        response_id=request_state.response_count,
        auth_state=rdf_flows.GrrMessage.AuthorizationState.AUTHENTICATED,
        type=rdf_flows.GrrMessage.Type.STATUS,
        payload=status)

    # Queue the response now
    self.queue_manager.QueueResponse(msg)
    self.QueueNotification(session_id=request_state.session_id)
777073
public function onPipeMessage($serv, $fromWorkerId, $message)
    {
        parent::onPipeMessage($serv, $fromWorkerId, $message);
        $data = unserialize($message);
        switch ($data['type']) {
            case Macro::MSG_TYPR_ASYN:
                $this->asynPoolManager->distribute($data['message']);
                break;
        }
    }
506276
public void send(ProducerPoolData<V> ppd) {
        if (logger.isDebugEnabled()) {
            logger.debug("send message: " + ppd);
        }
        if (sync) {
            Message[] messages = new Message[ppd.data.size()];
            int index = 0;
            for (V v : ppd.data) {
                messages[index] = serializer.toMessage(v);
                index++;
            }
            ByteBufferMessageSet bbms = new ByteBufferMessageSet(config.getCompressionCodec(), messages);
            ProducerRequest request = new ProducerRequest(ppd.topic, ppd.partition.partId, bbms);
            SyncProducer producer = syncProducers.get(ppd.partition.brokerId);
            if (producer == null) {
                throw new UnavailableProducerException("Producer pool has not been initialized correctly. " + "Sync Producer for broker "
                        + ppd.partition.brokerId + " does not exist in the pool");
            }
            producer.send(request.topic, request.partition, request.messages);
        } else {
            AsyncProducer<V> asyncProducer = asyncProducers.get(ppd.partition.brokerId);
            for (V v : ppd.data) {
                asyncProducer.send(ppd.topic, v, ppd.partition.partId);
            }
        }
    }