Groups messages by broker and preps them for transmission.

 @return [Array<MessagesForBroker>]
977900
def messages_for_brokers
      messages_for_broker_ids = {}
      @messages.each do |message|
        partition_id, broker_id = @message_conductor.destination(message.topic,
                                                                 message.key)

        # Create a nested hash to group messages by broker_id, topic, partition.
        messages_for_broker_ids[broker_id] ||= MessagesForBroker.new(broker_id)
        messages_for_broker_ids[broker_id].add(message, partition_id)
      end

      messages_for_broker_ids.values
    end
343795
def gather(self):
        
        if self.state in constants.RECEIVE_STATES:
            raise TypeError("Only new messages can be gathered.")
        if not self._message:
            raise ValueError("Message data already consumed.")
        try:
            raise self._response
        except TypeError:
            pass
        return [self]
84071
def messages(fp, key='@message'):
    
    for line in lines(fp):
        txt = line.rstrip('\n')
        yield {key: txt}
896953
public List<ChatMessage> adaptMessages(List<MessageReceived> messagesReceived) {

        List<ChatMessage> chatMessages = new ArrayList<>();

        if (messagesReceived != null) {
            for (MessageReceived msg : messagesReceived) {

                ChatMessage adaptedMessage = ChatMessage.builder().populate(msg).build();
                List<ChatMessageStatus> adaptedStatuses = adaptStatuses(msg.getConversationId(), msg.getMessageId(), msg.getStatusUpdate());
                for (ChatMessageStatus s : adaptedStatuses) {
                    adaptedMessage.addStatusUpdate(s);
                }
                chatMessages.add(adaptedMessage);
            }
        }

        return chatMessages;
    }
587512
public function getMessage()
    {
        $result = [];

        $messages = (isset($this->indexReverse[self::MESSAGE])) ? $this->indexReverse[self::MESSAGE] : [];

        foreach ($messages as $index) {
            $result[] = $this->getChild($index);
        }

        return $result;
    }
480159
def confirm_message
      cluster = Cluster.new

      case self.type
      when 'Elasticrawl::ParseJob'
        message = segment_list
      else
        message = []
      end

      message.push('Job configuration')
      message.push(self.job_desc)
      message.push('')
      message.push(cluster.cluster_desc)

      message.join("\n")
    end
1366902
protected SIBusMessage[] processMsgSet(SIMessageHandle[] msgHandles,
                                         TransactionCommon transaction,
                                         BifurcatedConsumerSessionImpl bifurcatedConsumer,
                                         boolean unlock,
                                         boolean delete,
                                         boolean read,
                                         boolean incrementLockCount) throws SIConnectionLostException, SIIncorrectCallException, SIResourceException, SIErrorException, SIMPMessageNotLockedException, SISessionDroppedException
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "processMsgSet",
        new Object[]{new Integer(hashCode()),
                     SIMPUtils.messageHandleArrayToString(msgHandles),
                     transaction,
                     bifurcatedConsumer,
                     new Boolean(unlock),
                     new Boolean(delete),
                     new Boolean(read),
                     new Boolean(incrementLockCount),
                     this});

    int numMsgs = msgHandles.length;
    LocalTransaction localTransaction = null;

    if (transaction != null && !transaction.isAlive())
    {
      SIIncorrectCallException e = new SIIncorrectCallException( nls.getFormattedMessage(
        "TRANSACTION_DELETE_USAGE_ERROR_CWSIP0778",
           new Object[] { consumerSession.getDestinationAddress() },
           null) );

      if (TraceComponent.isAnyTracingEnabled() && TraceComponent.isAnyTracingEnabled() && tc.isEventEnabled())
        SibTr.exception(tc, e);

      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
        SibTr.exit(tc, "processMsgSet", e);

      throw e;
    }

    // Create an array to hold any read messages
    SIBusMessage messages[] = null;
    if(read)
      messages = new SIBusMessage[numMsgs];

    SIMessageHandle missingMsgs[] = null;
    int missingMsgIndex = 0;
    int removedActiveMessages = 0;

    JsMessageWrapper messagesToUnlock[] = null;
    int unlockIndex = 0;

    JsMessageWrapper autoCommitMessage = null;

    try
    {
      synchronized(this)
      {
        int nextId = 0;

        // Search For each messageId in the LME, always moving forwards as both lists
        // are ordered
        for(nextId=0; nextId<numMsgs; nextId++)
        {
          if(msgHandles[nextId] != null)
          {
            //set the LME to start at the first entry
            LMEMessage lmeMessage = firstMsg;
            boolean msgFound = false;
            boolean expiredMsgFound;

            // Walk down the LME until we find the message or walk off the end
            while(lmeMessage != null)
            {
              expiredMsgFound = false;
              if( (lmeMessage.id == ((JsMessageHandle)msgHandles[nextId]).getSystemMessageValue())
                &&(lmeMessage.uuid.equals(((JsMessageHandle)msgHandles[nextId]).getSystemMessageSourceUuid()))
                &&(lmeMessage != currentUnlockedMessage) )
              {
                // We can only process messages that have not expired their lock
                if(!lmeMessage.lockExpired)
                {
                  // We also only process a message if it is still owned by the root consumer
                  // or we are processing a request from the owning bifurcated consumer
                  if((lmeMessage.owner == null) || (lmeMessage.owner == bifurcatedConsumer))
                  {
                    // Return the message to the caller if requested
                    if(read)
                    {
                      // Add the message item into the list to return.
                      messages[nextId] = lmeMessage.getJsMessage();

                      // Reading the message cancels the lock expiry
                      lmeMessage.expiryTime = 0;

                      // If this is being read by a bifurcated consumer we transfer the scope
                      // of its lock to them (unless we're going to delete it too, in which case
                      // there is no point)
                      if((bifurcatedConsumer != null) && !delete)
                      {
                        lmeMessage.owner = bifurcatedConsumer;
                      }
                    }

                    boolean markMsgAsMissing = false; // used to mark the message as missing when it is not in msgstore

                    // Delete the message if required
                    if (delete)
                    {
                      // Not all messages will still be in the message store
                      if(lmeMessage.isStored)
                      {
                        // If we weren't given a transaction we need to get one
                        if(transaction == null)
                        {
                          // If we have multiple messages to delete or we may perform extra
                          // messagestore operations as a result of the delete we'll need a
                          // local transaction
                          if((numMsgs > 1) ||
                             (!lmeMessage.message.isReference() &&
                              lmeMessage.message.getReportCOD() != null ||
                              lmeMessage.message.isRemoteGet()))
                          {
                            localTransaction = txManager.createLocalTransaction(!isRMQ());
                            transaction = localTransaction;
                          }
                          // Otherwise, we can get away with just re-using an autoCommitTran
                          // (but we can't use the LCP one as we don't hold any consumer level
                          // lock at this point, instead we use one owned by the LME as we hold the
                          // LME lock.
                          else
                          {
                            // Remember the the message to remove (to be done once the
                            // LME lock isn't held).
                            autoCommitMessage = lmeMessage.message;
                          }
                        }
                        else if (lmeMessage.message.isRemoteGet() && transaction.isAutoCommit())
                        {
                          // If an auto commit transaction was passed in and
                          // we have a remote ME then the transaction needs to be upgrated to a
                          // local transaction.
                          localTransaction = txManager.createLocalTransaction(!isRMQ());
                          transaction = localTransaction;
                        }

                        // If we have a true transaction we register for callbacks and
                        // remove the message. If we're using an autocommit transaction
                        // we can't remove it here under the LME lock as the remove may
                        // fail and re-drive the CD with the message which in turn will
                        // try to get the LCP lock. This breaks the locking hierarchy,
                        // AsynchLock->LCPLock->LMELock
                        if(autoCommitMessage == null)
                        {
                          try
                          {
                            removeMessageFromStore(lmeMessage.message,
                                                   transaction,
                                                   true); // true = decrement active message count (on commit)
                          }
                          catch (SIMPMessageNotLockedException e)
                          {
                            // No FFDC code needed
                            SibTr.exception(tc, e);
                            // See defect 387591
                            // We should only get this exception if the message was deleted via
                            // the controllable objects i.e. the admin console. If we are here for
                            // another reason that we have a real error.

                            // We don't need a special case here to decrement the active messages as the
                            // afterCompletion will still occur on a transaction with no workitems

                            markMsgAsMissing = true;
                          }
                        }
                      }
                      // Messages not in the MS need their count decrementing now, but not under the lock
                      else
                        removedActiveMessages++;

                      // Remove the reference to the message from the LME
                      removeMessage(lmeMessage);
                    }
                    // Or unlock it
                    else if(unlock)
                    {
                      // Not all messages will still be in the message store
                      if(lmeMessage.isStored)
                      {
                        // We can't unlock the message here as we hold the LME lock
                        // and the unlock could cause the message to be redelivered
                        // to the same consumer which will attempt to take the LCP
                        // lock - which breaks the lock hierarchy.

                        // Build an array of messages to unlock once the LME lock is
                        // released
                        if(messagesToUnlock == null)
                          messagesToUnlock = new JsMessageWrapper[numMsgs];

                        messagesToUnlock[unlockIndex++] = lmeMessage.message;
                      }
                      removeMessage(lmeMessage);

                      // If we're counting active messages decrement the count, but not under the lock
                      removedActiveMessages++;
                    }

                    if (!markMsgAsMissing) //only mark the msg as found if there wasn't a problem with it
                      msgFound = true;
                  }

                  // Break out the while now that we've processed the message
                  break;
                }
                // Otherwise, we simply remove the entry now that we've told someone
                else
                {
                  // Store the ref to the next msg as the remove destroys it
                  LMEMessage temp = lmeMessage.next;
                  removeMessage(lmeMessage);
                  // Next msg
                  lmeMessage = temp;
                  expiredMsgFound = true;
                }

              }
              if (!expiredMsgFound)
                lmeMessage = lmeMessage.next;
              if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                SibTr.debug(tc,"Checking msg for read and delete : "+lmeMessage);
            }
            if(!msgFound)
            {
              // We couldn't find the message, remember this
              if(missingMsgs == null)
                missingMsgs = new SIMessageHandle[numMsgs];

              missingMsgs[missingMsgIndex] = msgHandles[nextId];
              missingMsgIndex++;
            }
          }
          else
          {
            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
              SibTr.exit(tc, "processMsgSet", "SIIncorrectCallException one or more null SIMessageHandles");

            throw new SIIncorrectCallException(
              nls_cwsir.getFormattedMessage("LME_ERROR_CWSIR0132",
              null,
              null));
          }
        }
      }//synchronized

      // If we deferred removing the message we do it here now that the LME lock has
      // been released.
      if(autoCommitMessage != null)
      {
        try
        {
          removeMessageFromStore(autoCommitMessage,
                                 null,
                                 true); // true = decrement active message count
        }
        catch (SIMPMessageNotLockedException  e)
        {
          // No FFDC code needed
          SibTr.exception(tc, e);
          // As this should only occur when a msg has been deleted by the user then
          // we should have marked this message with been missing and so be processed
          // soon
        }

        // If we used an autoCommitTran we won't get any callbacks so we remember
        // we've removed another active message
        //removedActiveMessages++;
      }
      // if we created an internal transaction, commit it
      else if (localTransaction != null)
      {
        localTransaction.commit();
      }

      if(removedActiveMessages != 0)
        localConsumerPoint.removeActiveMessages(removedActiveMessages);

      //    If we deferred any unlocks until we didn't hold the LME lock, unlock them
      //    now
      if(unlockIndex != 0)
      {
        for(int i = 0; i < unlockIndex; i++)
        {
          try
          {
            unlockMessage(messagesToUnlock[i],incrementLockCount);
          }
          catch (SIMPMessageNotLockedException e)
          {
            // No FFDC code needed
            SibTr.exception(tc, e);
            // As this should only occur when a msg has been deleted by the user then
            // we should have marked this message with been missing and so be processed
            // soon
          }
        }
      }

      // If we haven't found all the message ids, then this is
      // an error and the SICoreMessageNotFound exception should
      // be thrown.
      if (missingMsgs != null)
      {
        // Crop the missing message array
        SIMessageHandle croppedMissingMsgs[] = new SIMessageHandle[missingMsgIndex];
        System.arraycopy(missingMsgs, 0, croppedMissingMsgs, 0, missingMsgIndex);

        SIMPMessageNotLockedException e =  new SIMPMessageNotLockedException (
          nls.getFormattedMessage(
                  "CORE_MESSAGE_NOT_FOUND_ERROR_CWSIP0173",
                  new Object[] {SIMPUtils.messageHandleArrayToString(croppedMissingMsgs),
                         localConsumerPoint.getConsumerManager().getDestination().getName(),
                         localConsumerPoint.getConsumerManager().getMessageProcessor().getMessagingEngineName() },
                  null),
         croppedMissingMsgs);

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
          SibTr.exit(tc, "processMsgSet", e);

        throw e;
      }
    }
    catch (SIResourceException e)
    {
      // MessageStoreException shouldn't occur so FFDC.
      FFDCFilter.processException(
        e,
        "com.ibm.ws.sib.processor.impl.AbstractLockedMessageEnumeration.processMsgSet",
        "1:1824:1.154.3.1",
        this);

      SibTr.exception(tc, e);
      SibTr.error(tc, "INTERNAL_MESSAGING_ERROR_CWSIP0002",
        new Object[] {
          "com.ibm.ws.sib.processor.impl.AbstractLockedMessageEnumeration",
          "1:1831:1.154.3.1",
          e });

      // If we created a transaction, try to roll it back before we
      // exit
      if(localTransaction != null)
      {
        localTransaction.rollback();
      }

      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
        SibTr.exit(tc, "processMsgSet", e);

      throw e;
    }

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "processMsgSet", new Object[] {messages, this});

    return messages;
  }
1241551
def full_messages
      inject([]) do |m, kv|
        att, errors = *kv
        errors.each {|e| m << "#{e}"}
        m
      end
    end
658347
function() {
    _cbt =  require('./cbt')();
    _options = options || {};

    // apply defaults
    // the min bytes of an incoming reply
    _options.minBytes = _options.minBytes || 1;

    // the max bytes of an incoming reply
    _options.maxBytes = _options.maxBytes || 1024 * 1024;

    // the group the client will join when using `.connect()`
    _options.group = _options.group || 'kafkaesqueGroup';

    // the clientID when connecting to kafka
    _options.clientId = _options.clientId || 'kafkaesque' + Math.floor(Math.random() * 100000000);

    // the array of brokers
    _options.brokers = _options.brokers || [{host: 'localhost', port: 9092}];

    // the amount of time it should take for this clients session within the group
    // to timeout
    _options.sessionTimeout = _options.sessionTimeout || 6000;

    // the amount of time to take to emit a heartbeat msg
    _options.heartbeat = _options.heartbeat || 2500;

    // the default amount of time that the kafka broker shoudl wait to send
    // a reply to a fetch request if the fetch reply is smaller than the minBytes
    _options.maxWait = _options.maxWait || 5000;

    _brokers = {};
    _topics = {};
    _groupMemberId = '';
    _groupGeneration = 0;
    _.each(_options.brokers, function(broker) {
      broker = {
        host: broker.host,
        port: broker.port,
        maxBytes: _options.maxBytes,
        minBytes: _options.minBytes,
        clientId: _options.clientId
      };

      _brokers[_makeBrokerKey(broker)] = api(broker);
    });

    _metaBroker = _brokers[_makeBrokerKey(_options.brokers[0])];
  }
690795
private List<ProxyMessage> createProxyMessages(final Page<String> messages) {
        final List<ProxyMessage> proxyMsgs = new ArrayList<>(messages.getNumberOfElements());

        Long idx = messages.getNumber() * ((long) messages.getSize());
        for (final String msg : messages) {
            idx++;
            final ProxyMessage proxyMsg = new ProxyMessage();
            proxyMsg.setMessage(msg);
            proxyMsg.setId(String.valueOf(idx));
            proxyMsgs.add(proxyMsg);
        }

        if (messages.getTotalElements() == 1L && StringUtils.isEmpty(proxyMsgs.get(0).getMessage())) {
            proxyMsgs.get(0).setMessage(noMessageText);
        }

        return proxyMsgs;
    }
1307107
public function getMessages()
    {
        $messages = parent::getMessages();

        foreach ($this->pages as $page) {
            $messages = array_merge($messages, $page->getMessages());
        }

        return $messages;
    }
1051166
def message_create(originator, recipients, body, params={})
      # Convert an array of recipients to a comma-separated string.
      recipients = recipients.join(',') if recipients.kind_of?(Array)

      Message.new(request(
        :post,
        'messages',
        params.merge({
          :originator => originator.to_s,
          :body       => body.to_s,
          :recipients => recipients })))
    end
415095
def prepare_dispatches(cls, message, recipients=None):
        
        return Dispatch.create(message, recipients or cls.get_subscribers())
926268
def to_s
      message = ''
      @messages.keys.sort.each do |timestamp|
        message << "#{Time.at(timestamp / 1000)}"
        message << " #{@messages[timestamp]['severity']}: #{@messages[timestamp]['user']}:"
        message << " #{@messages[timestamp]['short-description']}"
        unless @messages[timestamp]['short-description'].eql?(@messages[timestamp]['long-description'])
          message << "; #{@messages[timestamp]['long-description']}"
        end
        message << "\n"
      end
      message
    end
1076585
private Map<String, SortedSet<Partition>> getZKTopicPartitionInfo(Map<Integer, Broker> allBrokers) {
        final Map<String, SortedSet<Partition>> brokerPartitionsPerTopic = new HashMap<String, SortedSet<Partition>>();
        ZkUtils.makeSurePersistentPathExists(zkClient, ZkUtils.BrokerTopicsPath);
        List<String> topics = ZkUtils.getChildrenParentMayNotExist(zkClient, ZkUtils.BrokerTopicsPath);
        for (String topic : topics) {
            // find the number of broker partitions registered for this topic
            String brokerTopicPath = ZkUtils.BrokerTopicsPath + "/" + topic;
            List<String> brokerList = ZkUtils.getChildrenParentMayNotExist(zkClient, brokerTopicPath);
            //
            final SortedSet<Partition> sortedBrokerPartitions = new TreeSet<Partition>();
            final Set<Integer> existBids = new HashSet<Integer>();
            for (String bid : brokerList) {
                final int ibid = Integer.parseInt(bid);
                final String numPath = brokerTopicPath + "/" + bid;
                final Integer numPartition = Integer.valueOf(ZkUtils.readData(zkClient, numPath));
                for (int i = 0; i < numPartition.intValue(); i++) {
                    sortedBrokerPartitions.add(new Partition(ibid, i));
                }
                existBids.add(ibid);
            }
            // add all brokers after topic created
            for(Integer bid:allBrokers.keySet()){
                if(!existBids.contains(bid)){
                    sortedBrokerPartitions.add(new Partition(bid,0));// this broker run after topic created
                }
            }
            logger.debug("Broker ids and # of partitions on each for topic: " + topic + " = " + sortedBrokerPartitions);
            brokerPartitionsPerTopic.put(topic, sortedBrokerPartitions);
        }
        return brokerPartitionsPerTopic;
    }
540865
def open_receiver_class
      # Define association for receiver model
      [receivers_class(@@options[:receivers])].flatten.each do |base|
        base.class_eval do
          has_many :deliveries, class_name: 'SimpleNotifications::Delivery', as: :receiver
          has_many :received_notifications, through: :deliveries, source: :simple_notification
        end
      end
    end
1146142
def message(self, msg):
        
        for broker in self.message_brokers:
            try:
                broker(msg)
            except Exception as exc:
                utils.error(exc)