src/STOMP_EncDec.cc

/******************************************************************************
* Copyright (c) 2000-2019 Ericsson Telecom AB AB
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v2.0
* which accompanies this distribution, and is available at
* https://www.eclipse.org/org/documents/epl-2.0/EPL-2.0.html
*
* Contributors:
*  Bence Janos Szabo - initial implementation
******************************************************************************/
//
//  File:               STOMP_EncDec.cc
//  Rev:                R1B           
//  Prodnr:             

#include "STOMP_Types.hh"

namespace STOMP__Types {

void encode_utf8_stomp(TTCN_Buffer& stream, const UNIVERSAL_CHARSTRING &value, const bool escape);
void decode_utf8_stomp(UNIVERSAL_CHARSTRING &value, const bool escape);

INTEGER f__STOMP__dec(const OCTETSTRING &str, STOMPFrame &msg)
{
  if (TTCN_Logger::log_this_event(TTCN_DEBUG)) {
    TTCN_Logger::begin_event(TTCN_DEBUG);
    TTCN_Logger::log_event("Decoding STOMP message: ");
    str.log();
    TTCN_Logger::end_event();
  }

  msg.clean_up();

  TTCN_Buffer stream;
  const int length = str.lengthof();
  int prev_pos = 0;
  int pos = 0;
  int delta;

  // Find the [\r]\n to identify the frame type
  while (pos < length) {
    if (str[pos].get_octet() == '\n') {
      delta = 1;
      break;
    } else if (str[pos].get_octet() == '\r' && pos + 1 < length && str[pos+1].get_octet() == '\n') {
      delta = 2;
      break;
    }
    pos++;
  }
  
  CHARSTRING frame_type(pos, reinterpret_cast<const char*>((const unsigned char*)str));
  pos += delta;

  bool escape = true;

  if (frame_type == "SEND") {
    msg.command() = Command::SEND;
  } else if (frame_type == "SUBSCRIBE") {
    msg.command() = Command::SUBSCRIBE;
  } else if (frame_type == "UNSUBSCRIBE") {
    msg.command() = Command::UNSUBSCRIBE;
  } else if (frame_type == "BEGIN") {
    msg.command() = Command::BEGIN;
  } else if (frame_type == "COMMIT") {
    msg.command() = Command::COMMIT;
  } else if (frame_type == "ABORT") {
    msg.command() = Command::ABORT;
  } else if (frame_type == "ACK") {
    msg.command() = Command::ACK;
  } else if (frame_type == "NACK") {
    msg.command() = Command::NACK;
  } else if (frame_type == "DISCONNECT") {
    msg.command() = Command::DISCONNECT;
  } else if (frame_type == "CONNECTED") {
    msg.command() = Command::CONNECTED;
    escape = false;
  } else if (frame_type == "CONNECT") {
    msg.command() = Command::CONNECT;
    escape = false;
  } else if (frame_type == "STOMP") {
    msg.command() = Command::STOMP;
  } else if (frame_type == "MESSAGE") {
    msg.command() = Command::MESSAGE;
  } else if (frame_type == "RECEIPT") {
    msg.command() = Command::RECEIPT;
  } else if (frame_type == "ERROR") {
    msg.command() = Command::ERROR_;
  } else {
    TTCN_error("f_STOMP_dec error: Unexpected frame type received: %s.", (const char*)frame_type);
  }

  size_t index = 0;
  bool found_payload_length = false;
  int payload_length;
  while (true) {
    prev_pos = pos;
    while (pos < length) {
      if (str[pos].get_octet() == '\n') {
        delta = 1;
        break;
      } else if (str[pos].get_octet() == '\r' && pos + 1 < length && str[pos+1].get_octet() == '\n') {
        delta = 2;
        break;
      }
      pos++;
    }

    // There are some headers
    if (pos - prev_pos > 0) {
      UNIVERSAL_CHARSTRING header;
      header.decode_utf8(pos-prev_pos, &((const unsigned char*)str)[prev_pos]);
      const int header_len = header.lengthof();
      for (int i = 0; i < header_len; i++) {
        if (header[i].get_uchar().is_char() && header[i].get_uchar().uc_cell == ':') {
          msg.headers()[index].header__name() = UNIVERSAL_CHARSTRING(i, (const universal_char*)header);
          msg.headers()[index].header__value() = UNIVERSAL_CHARSTRING(header_len-i-1, ((const universal_char*)header)+i+1);
          decode_utf8_stomp(msg.headers()[index].header__name(), escape);
          decode_utf8_stomp(msg.headers()[index].header__value(), escape);
          if (found_payload_length == false && msg.headers()[index].header__name() == "content-length") {
            found_payload_length = true;
            TTCN_Buffer tmp_buff;
            msg.headers()[index].header__value().encode_utf8(tmp_buff);
            CHARSTRING cs(tmp_buff.get_len(), (const char*)tmp_buff.get_data());
            payload_length = str2int(cs);
          }
          index++;
          break;
        }
      }
    } else {
      pos += delta;
      break;
    }
    pos += delta;
  }

  if (msg.headers().is_bound() == false) {
    msg.headers().set_size(0);
  }

  prev_pos = pos;
  if (found_payload_length == true) {
    pos += payload_length;
    if (pos >= length) {
      TTCN_error("f_STOMP_dec error: Not enough bytes in the bytestream.");
    }
  } else {
    // Read until closing NULL
    const unsigned char* payload = ((const unsigned char*)str)+pos;
    while (pos < length && *payload != 0) {
      payload++;
      pos++;
    }
  }

  if (pos-prev_pos > 0) {
    msg.payload() = OCTETSTRING(pos-prev_pos, ((const unsigned char*)str)+prev_pos);
  } else {
    msg.payload() = OMIT_VALUE;
  }

  // Skip NULL
  pos++;
  // read remaining [\r]\n
  while (pos < length) {
    if (str[pos].get_octet() == '\n') {
      pos += 1;
    } else if (str[pos].get_octet() == '\r') {
      if (pos + 1 < length ) {
        if (str[pos+1].get_octet() == '\n') {
          pos += 2;
        } else {
          TTCN_error("f_STOMP_dec error: Extra non endline characters after the end of frame: %c.", str[pos+1].get_octet());
        }
      } else {
        TTCN_error("f_STOMP_dec error: \r character found but no \n.");
      }
    } else {
      TTCN_error("f_STOMP_dec error: Extra non endline characters after the end of frame: %c.", str[pos].get_octet());
    }
  }

  if (TTCN_Logger::log_this_event(TTCN_DEBUG)) {
    TTCN_Logger::begin_event(TTCN_DEBUG);
    TTCN_Logger::log_event("Decoded STOMP message: ");
    msg.log();
    TTCN_Logger::end_event();
  }

  return 0;
}


INTEGER f__STOMP__enc(const STOMPFrame &msg, OCTETSTRING &str)
{
  static const char EOL = '\n';
  TTCN_Buffer stream;

  if (TTCN_Logger::log_this_event(TTCN_DEBUG)) {
    TTCN_Logger::begin_event(TTCN_DEBUG);
    TTCN_Logger::log_event("Encoding STOMP message: ");
    msg.log();
    TTCN_Logger::end_event();
  }

  bool escape = true;

  // Commands
  if (msg.command().is_bound()) {
    switch (msg.command()) {
      case Command::SEND:
        stream.put_cs("SEND");
        break;
      case Command::SUBSCRIBE:
        stream.put_cs("SUBSCRIBE");
        break;
      case Command::UNSUBSCRIBE:
        stream.put_cs("UNSUBSCRIBE");
        break;
      case Command::BEGIN:
        stream.put_cs("BEGIN");
        break;
      case Command::COMMIT:
        stream.put_cs("COMMIT");
        break;
      case Command::ABORT:
        stream.put_cs("ABORT");
        break;
      case Command::ACK:
        stream.put_cs("ACK");
        break;
      case Command::NACK:
        stream.put_cs("NACK");
        break;
      case Command::DISCONNECT:
        stream.put_cs("DISCONNECT");
        break;
      case Command::CONNECTED:
        stream.put_cs("CONNECTED");
        escape = false;
        break;
      case Command::CONNECT:
        stream.put_cs("CONNECT");
        escape = false;
        break;
      case Command::STOMP:
        stream.put_cs("STOMP");
        break;
      case Command::MESSAGE:
        stream.put_cs("MESSAGE");
        break;
      case Command::RECEIPT:
        stream.put_cs("RECEIPT");
        break;
      case Command::ERROR_:
        stream.put_cs("ERROR");
        break;
      default:
        TTCN_error("f_STOMP_enc error: Unrecognized command.");
    }
  } else {
    TTCN_error("f_STOMP_enc error: Unbound command field.");
  }

  stream.put_c(EOL); // End of line signals that the headers are coming

  // We put in automatically the content-length if it is not present 
  // and the payload field is present.
  bool found_content_length = false;

  // Header names and values with escaping if needed
  if (msg.headers().is_bound()) {
    const int nof_headers = msg.headers().size_of();
    for (int i = 0; i < nof_headers; i++) {
      if (msg.headers()[i].header__name() == "content-length") {
        found_content_length = true;
      }
      encode_utf8_stomp(stream, msg.headers()[i].header__name(), escape);
      stream.put_c(':');
      encode_utf8_stomp(stream, msg.headers()[i].header__value(), escape);
      stream.put_c(EOL); // End of line separates the headers from each other
    }
  } else {
    TTCN_error("f_STOMP_enc error: Unbound headers field.");
  }

  if (msg.payload().is_bound()) {
    if (msg.payload().ispresent()) {
      // Automatically add content-length header
      if (found_content_length == false) {
        encode_utf8_stomp(stream, "content-length", escape);
        stream.put_c(':');
        CHARSTRING cs = int2str(msg.payload()().lengthof());
        encode_utf8_stomp(stream, cs, escape);
        stream.put_c(EOL);
      }

      stream.put_c(EOL); // An empty line signals that the payload may come
      stream.put_string(msg.payload());
    } else {
      stream.put_c(EOL); // An empty line signals that the payload may come, but it wont.
    }
  } else {
    TTCN_error("f_STOMP_enc error: Unbound payload field.");
  }

  stream.put_c(0); // End of frame

  if (TTCN_Logger::log_this_event(TTCN_DEBUG)) {
    TTCN_Logger::begin_event(TTCN_DEBUG);
    TTCN_Logger::log_event("Encoded STOMP message: ");
    stream.log();
    TTCN_Logger::end_event();
  }

  stream.get_string(str);
  return 0;
}


void encode_utf8_stomp(TTCN_Buffer& stream, const UNIVERSAL_CHARSTRING &value, const bool escape) {
  if (escape) {
    TTCN_Buffer tmp_buff;
    value.encode_utf8(tmp_buff);
    const unsigned char* ustr = tmp_buff.get_data();
    const size_t pos = tmp_buff.get_len();
    for (size_t j = 0; j < pos; j++) {
      switch (ustr[j]) {
        case '\r':
          stream.put_cs("\\r");
          break;
        case '\n':
          stream.put_cs("\\n");
          break;
        case ':':
          stream.put_cs("\\c");
          break;
        case '\\':
          stream.put_cs("\\\\");
          break;
        default:
          stream.put_c(ustr[j]);
          break;
      }
    }
  } else {
    value.encode_utf8(stream);
  }
}

void decode_utf8_stomp(UNIVERSAL_CHARSTRING &value, const bool escape) {
  if (escape) {
    UNIVERSAL_CHARSTRING result = "";
    const int length = value.lengthof();
    for (int i = 0; i < length; i++) {
      if (i+1 < length && value[i].get_uchar().is_char() && value[i+1].get_uchar().is_char()) {
        if (value[i].get_uchar().uc_cell == '\\') {
          i++;
          const universal_char& uchar_value = value[i].get_uchar();
          switch (uchar_value.uc_cell) {
            case 'r':
              result = result + "\r";
              break;
            case 'n':
              result = result + "\n";
              break;
            case 'c':
              result = result + ":";
              break;
            case '\\':
              result = result + "\\";
              break;
            default:
              TTCN_error("f_STOMP_dec error: Undefined escape sequence: \\%c", uchar_value.uc_cell);
              break;
          }
        } else {
          result = result + value[i].get_uchar();
        }
      } else {
        result = result + value[i].get_uchar();
      }
    }
    value = result;
  }
}


}