code: @Override
      public void execute(Tuple input)
      {
          // extract message from Tuple
          this.executingTuple = input;
  
          StreamMessage message = null;
          boolean messageGet = false;
          if (input.contains(FieldName.MESSAGE_VALUE) == true)
          {
              Object obj = input.getValueByField(FieldName.MESSAGE_VALUE);
              if (obj instanceof StreamMessage)
              {
                  message = (StreamMessage) obj;
                  messageGet = true;
              }
          }
  
          if (message == null)
          {
              message = new StreamMessage();
          }
  
          for (String field : input.getFields())
          {
              if (messageGet && FieldName.MESSAGE_VALUE.equals(field))
              {
                  continue;
              }
  
              if (messageGet && FieldName.MESSAGE_KEY.equals(field))
              {
                  Object obj = input.getValueByField(field);
                  if (message.getHeader() != null)
                  {
                      message.getHeader().setMessageKey(obj.toString());
                  }
                  continue;
              }
  
              Object obj = input.getValueByField(field);
              message.addField(field, obj);
          }
  
          try
          {
              onMessage(message);
          }
          finally
          {
              this.executingTuple = null;
          }
      }
docs: {@inheritDoc}