code: @Override
      public void nextTuple()
      {
          Message message = null;
  
          try
          {
              message = this.connection.receive(this.receiveWait, TimeUnit.SECONDS);
          }
          catch (Exception ex)
          {
              String msg = "Receive failed. Retry receive.";
              logger.warn(msg, ex);
              this.connection.resume();
          }
  
          // 取得できなかった場合は処理を終了する。
          if (message == null)
          {
              return;
          }
  
          String topic = message.getTopic();
          String payload = new String(message.getPayload(), Charset.forName("UTF-8"));
          acromusashi.stream.entity.StreamMessage sendMessage = createMessage(topic, payload);
  
          // 即応答を返す設定がされているか、またはQoSレベルが0の場合はAnchorを用いずに即Ackを返したうえでTupleを流す。
          // 上記以外の場合はランダムで生成したUUIDをAnchorとしてTupleを流す。
          if (this.immidiateAck == true || this.qos == QoS.AT_MOST_ONCE)
          {
              message.ack();
              getCollector().emit(new Values(topic, sendMessage));
          }
          else
          {
              String messageId = UUID.randomUUID().toString();
              this.ackWaitMap.put(messageId, message);
              getCollector().emit(new Values(topic, sendMessage), messageId);
          }
      }
docs: {@inheritDoc}