/*
 * Copyright (c) 2015, 2016, Oracle and/or its affiliates. All rights reserved.
 *
 * This code is licensed under the terms of the GPLv2
 * <http://www.gnu.org/licenses/old-licenses/gpl-2.0.html>, like most
 * MySQL Connectors. There are special exceptions to the terms and
 * conditions of the GPLv2 as it is applied to this software, see the
 * FLOSS License Exception
 * <http://www.mysql.com/about/legal/licensing/foss-exception.html>.
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published
 * by the Free Software Foundation; version 2 of the License.
 *
 * This program is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
 * or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
 * for more details.
 *
 * You should have received a copy of the GNU General Public License along
 * with this program; if not, write to the Free Software Foundation, Inc.,
 * 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA
 */

/*
  Implementation of mysqlx protocol API
  =====================================
*/

// Note: on Windows this includes windows.h

#include <mysql/cdk/foundation/common.h>

/*
  Note: On Windows the INIT_ONCE structure was added only in later
  releases but can be available in previous ones via service packs.
*/

#ifdef _WIN32
  static INIT_ONCE log_handler_once;
  #define EXECUTE_ONCE(A, B) InitOnceExecuteOnce(A, B, NULL, NULL)
#else
  #include <pthread.h>
  pthread_once_t log_handler_once = PTHREAD_ONCE_INIT;
  #define EXECUTE_ONCE(A, B) pthread_once(A, B)
#endif

#include "protocol.h"

PUSH_SYS_WARNINGS
#include <memory.h> // for memcpy
POP_SYS_WARNINGS


#ifdef DEBUG_PROTOBUF

PUSH_SYS_WARNINGS
#include <iostream>
POP_SYS_WARNINGS

#define MSG_CLIENT_name_client(MSG,N,C) case msg_type::cli_##N: return #MSG;
#define MSG_SERVER_name_client(MSG,N,C)
#define MSG_CLIENT_name_server(MSG,N,C)
#define MSG_SERVER_name_server(MSG,N,C) case msg_type::N: return #MSG;

namespace cdk {
namespace protocol {
namespace mysqlx {

const char* msg_type_name(Protocol_side side, msg_type_t type)
{
  switch (side)
  {
  case SERVER:
    switch (type)
    {
      MSG_LIST(name_server)
      default: return "<unknown msg type>";
    }
  case CLIENT:
    switch (type)
    {
      MSG_LIST(name_client)
      default: return "<unknown msg type>";
    }
  }
  return NULL;
}

}}}  // cdk::protocol::mysqlx

#endif


IMPL_PLAIN(cdk::protocol::mysqlx::Protocol);
IMPL_PLAIN(cdk::protocol::mysqlx::Protocol_server);


using namespace cdk::foundation;
using namespace google::protobuf;
using namespace cdk::protocol::mysqlx;


namespace cdk {
namespace protocol {
namespace mysqlx {


/*
  Protobuf log handler initialization.
*/

static void log_handler(LogLevel level, const char* filename, int line, const std::string& message);

#ifdef _WIN32
BOOL CALLBACK log_handler_init( PINIT_ONCE InitOnce, PVOID Parameter, PVOID *lpContext)
{
  SetLogHandler(log_handler);
  return TRUE;
}
#else
static void log_handler_init()
{
  SetLogHandler(log_handler);
}
#endif


/*
  Base protocol implementation
  ============================
*/


Protocol_impl::Protocol_impl(Protocol::Stream *str, Protocol_side side)
  : m_str(str), m_side(side)
  , m_msg_state(PAYLOAD)
  , m_msg_size(0)
{
  EXECUTE_ONCE(&log_handler_once, &log_handler_init);

  // Allocate initial I/O buffers

  m_wr_size= m_rd_size= 512;
  m_rd_buf= (byte*)malloc(m_rd_size);
  m_wr_buf= (byte*)malloc(m_wr_size);

  if (!m_rd_buf)
    throw_error("Could not allocate initial input buffer");

  if (!m_wr_buf)
    throw_error("Could not allocate initial output buffer");
}

Protocol_impl::~Protocol_impl()
{
  free(m_rd_buf);
  free(m_wr_buf);
  delete m_str;
}


class Invalid_msg_error : public Error_class<Invalid_msg_error>
{
  unsigned m_state;
  msg_type_t m_type;

public:

  Invalid_msg_error(msg_type_t type, unsigned state)
    : Error_base(NULL, cdkerrc::generic_error),
      m_state(state), m_type(type)
  {}

private:

  void do_describe(std::ostream &out) const
  {
    out <<"Message of type " <<(unsigned)m_type
        <<" is not valid in protocol state " <<m_state;
  }
};


Protocol::Op& Protocol_impl::snd_start(Message &msg, msg_type_t msg_type)
{

#ifdef DEBUG_PROTOBUF

  using std::cerr;
  using std::endl;

  cerr << endl;
  cerr << ">>>> Seding message >>>>" << endl;
  cerr << "of type " << msg_type << ": "
      << msg_type_name(CLIENT, msg_type) << endl;
  cerr << msg.DebugString();
  cerr << ">>>>" << endl << endl;

#endif

  //First delete completed OP, so that if Snd_op() throws exception m_snd_op
  //will not point to old OP.
  m_snd_op.reset();
  m_snd_op.reset(new Op_snd(*this, msg_type, msg));
  return *m_snd_op;
}


/*
  Helper function which creates protobuf message object of type
  indicated by msg_type identifier. Interpretation of msg_type_t
  values depends on whether we look at server-side or client-side
  messages.
*/

Message* mk_message(Protocol_side side, msg_type_t msg_type)
{
    /*
      The case clauses for switch statements that generate
      messages of appropriate type are again generated using
      MSG_SERVER() and MSG_CLIENT() entries in MSG_LIST()
      macro (see protocol/mysqlx.h for definition and description).
    */

#define MSG_CLIENT_MK_CLI(MSG,N,C) \
  case msg_type::cli_##N: return new MSG();
#define MSG_SERVER_MK_CLI(MSG,N,C)
#define MSG_SERVER_MK_SRV(MSG,N,C) \
  case msg_type::N: return new MSG();
#define MSG_CLIENT_MK_SRV(MSG,N,C)

    switch (side)
    {
    case SERVER:
      switch (msg_type)
      {
        MSG_LIST(MK_SRV)
        // TODO: proper error
        default: THROW("unknown server message type");
      }; break;

    case CLIENT:
      switch (msg_type)
      {
        MSG_LIST(MK_CLI)
        default: THROW("unknown server message type");
      }; break;

    default: THROW("unknown protocol side");
    }
}



/*
  Protobuf error logger
  =====================

  This method is registered with Protobuf and called when warning/error/fatal
  situation occurs in Protobuf (such as parsing error etc).
*/

static void log_handler(LogLevel level, const char* filename, int line, const std::string& message)
{
  switch(level)
  {
    case LOGLEVEL_FATAL:
    case LOGLEVEL_ERROR:
      /*
        With this code the error description is:

          MMM: Protobuf error (cdk:NNN)

        where MMM is the message and NNN is the protbuf error code.

        TODO: Change description to:

          Protobuf error: MMM (cdk:NNN)
      */
      throw_error(cdkerrc::protobuf_error, message);

    case LOGLEVEL_WARNING:
    case LOGLEVEL_INFO:
    default:
    {
      // just ignore for now
      // TODO: this could be used for logging in the future
    }
  }
}


/*
  Writing/reading message frames
  ==============================
*/


void Protocol_impl::write_msg(msg_type_t msg_type, Message &msg)
{
  if (m_wr_op)
    THROW("Can't write message while another one is written");

  msg_size_t net_size = static_cast<unsigned>(msg.ByteSize()) + 1;

  if (!resize_buf(CLIENT, header_length + net_size))
    THROW("Not enough memory for output buffer");

  // Construct message header

  HTONSIZE(net_size);
  memcpy((void*)m_wr_buf, (const void*)&net_size, sizeof(net_size));
  m_wr_buf[header_length - 1] = (byte)msg_type;

  // Convert net_size back to original endian before using it later

  NTOHSIZE(net_size);

  // Serialize message

  assert(m_wr_size < (size_t)std::numeric_limits<int>::max());

  if (!msg.SerializeToArray((void*)(m_wr_buf + header_length),
                            (int)(m_wr_size - header_length)))
    throw_error(cdkerrc::protobuf_error, "Serialization error!");

  // Create write operation to send message payload

  m_wr_op.reset(m_str->write(buffers(m_wr_buf, net_size + header_length - 1)));
}


bool Protocol_impl::wr_cont()
{
  if (!m_wr_op)
    return true;

  if (!m_wr_op->cont())
    return false;

  m_wr_op.reset();
  return true;
}


void Protocol_impl::wr_wait()
{
  if (m_wr_op)
  {
    m_wr_op->wait();
    m_wr_op.reset();
  }
}


void Protocol_impl::read_header()
{
  if (HEADER == m_msg_state)
    return;

  if (m_rd_op)
    THROW("can't read header when reading payload is not completed");

  m_rd_op.reset(m_str->read(buffers(m_rd_buf,4)));
  m_msg_state= HEADER;
}

void Protocol_impl::read_payload()
{
  if (PAYLOAD == m_msg_state)
    return;

  if (HEADER != m_msg_state)
    THROW("payload can be read only after header");

  if (m_rd_op)
    THROW("can't read payload when reading header is not completed");

  if (!resize_buf(SERVER, m_msg_size))
      THROW("Not enough memory for input buffer");

  if (m_msg_size > 0)
    m_rd_op.reset(m_str->read(buffers(m_rd_buf, m_msg_size)));
  m_msg_state= PAYLOAD;
}


bool Protocol_impl::rd_cont()
{
  if (!m_rd_op)
    return true;

  if (!m_rd_op->cont())
    return false;

  m_rd_op.reset();

  if (PAYLOAD == m_msg_state)
    return true;

  rd_process();

  return true;
}


void Protocol_impl::rd_wait()
{
  if (m_rd_op)
  {
    m_rd_op->wait();
    m_rd_op.reset();

    if (PAYLOAD == m_msg_state)
      return;

    rd_process();
  }
}


bool Protocol_impl::resize_buf(Protocol_side side, size_t requested_size)
{
  byte*  &buf= (side == SERVER ? m_rd_buf : m_wr_buf);
  size_t &buf_size= (side == SERVER ? m_rd_size : m_wr_size);

  if (requested_size < buf_size)
    return true;

  // Note that since requested_size >= buf_size, the buffer size is
  // at least doubled here.

  size_t new_size= buf_size + requested_size;
  byte *ptr= (byte*) realloc(buf, new_size);

  // If allocating buffer with margin failed, try allocating
  // exact required amount.

  if (!ptr)
  {
    new_size= requested_size;
    ptr= (byte*) realloc(buf, new_size);
  }

  if (!ptr)
    return false;

  buf_size = new_size;
  buf= ptr;

  return true;
}


void Protocol_impl::rd_process()
{
  m_msg_size= *(msg_size_t*)m_rd_buf;
  NTOHSIZE(m_msg_size);
  assert(m_msg_size > 0);
  m_msg_size--;

  m_rd_op.reset(m_str->read(buffers(m_rd_buf, 1)));
  m_rd_op->wait();
  m_rd_op.reset();
  m_msg_type= m_rd_buf[0];
}


/*
  Processing incoming messages
  ============================
*/


/*
  Main logic of receive operation which reads messages and passes data
  to message processor.

  The async flag tells whether a call to do_read_msg() should proceed until
  complete message is read and processed (if async is false) or whether it
  can return before completing the operation (if async is true) in which case
  another call to do_read_msg() will push it further.

  Returns false if the operation is not completed yet.
*/


bool Op_rcv::do_read_msg(bool async)
{
  while (!m_completed)
  {
    switch (m_stage)
    {
      case HEADER:
      {
        // first make sure header is read.

        if (!async)
          m_proto.rd_wait();
        else if (!m_proto.rd_cont())
          return false;

        m_msg_type = m_proto.m_msg_type;

        // Check whether message is expected and whether operation
        // should stop here.

        Next_msg next = next_msg(m_msg_type);

        if (STOP == next)
        {
          finish(false);  // false = do not read next message
          if (async)
            return true;
          continue;
        }

        /*
          Call message_begin(). The Boolean flag passed to this method
          tells if this message is expected. The processor can alter the
          flag to change the way message is processed below.
        */

        bool flag = (EXPECTED == next);

        if (!m_error && m_prc)
        {
          try
          {
            m_read_window= m_prc->message_begin(m_msg_type, flag);
            m_call_message_end = true;
          }
          catch (...)
          {
            save_error();
            m_call_message_end = false;
          }
        }

        // Interpret disposition how to handle the message

        m_skip = false;
        if (UNEXPECTED == next)
        {
          /*
            If message was not expected, the initial value of the flag was
            false. If the processor has not changed it, we throw unexpected
            message error here. Otherwise, the processor told us to skip this
            unexpected message and continue processing.
          */
          if (!flag)
          {
            finish(false);
            THROW("unexpected message");
          }
          m_skip = true;
        }
        else if (!flag)
        {
          /*
            If message was expected, the initial value of the flag was true.
            If the processor has changed it then it means it does not want
            to see this message and we can skip it.
          */
          m_skip = true;
        }

        // Start reading payload

        m_proto.read_payload();
        m_stage = PAYLOAD;

        // fall-through to payload processing phase
      }

      case PAYLOAD:
      {
        // Ensure that payload is read

        if (!async)
          m_proto.rd_wait();
        else if (!m_proto.rd_cont())
          return false;

        m_msg_size = m_proto.m_msg_size;

        // process the payload

        if (m_prc && !m_error)
          process_payload();

        /*
          call message_end() - the return value can tell us to stop
          processing here regardless of the current state.
        */

        bool stop = false;
        if (m_prc && m_call_message_end)
        {
          try
          {
            stop = !m_prc->message_end();
          }
          catch(...)
          {
            save_error();
          }
        }

        m_stage = DONE;

        /*
          Pass true to finish() to read next message if process_next()
          tells us so and the processor has not interrupted the processing.

          Note: it is important to always call process_next() because derived
          classes rely on it being called after processing each message to
          do final chores.
        */

        bool done = finish(process_next() && !stop);

        if (async)
          return done;
      }
      break;

    case DONE:
      return true;
    }
  }

  return true;
}


bool Op_rcv::do_cont()
{
  return do_read_msg(true);
}

void Op_rcv::do_wait()
{
  do_read_msg(false);
}


/*
  Process received message payload. After parsing, the parsed message
  is processed with process_msg() method which can be overridden by
  derived classes.
*/

void Op_rcv::process_payload()
{
  assert(m_prc);
  assert(PAYLOAD == m_stage);

  // Send raw message bytes to m_proto if requested (m_read_window > 0).

  try {

    byte *cur_pos = m_proto.m_rd_buf;
    byte *end_pos = cur_pos + m_msg_size;

    /*
      Note: read_header() checks if message fits into the buffer and
      throws error if this is not the case.
    */

    assert(m_msg_size <= m_proto.m_rd_size);

    while (cur_pos < end_pos && m_read_window)
    {
      size_t new_window = m_prc->message_data(bytes(cur_pos,
                                 cur_pos + m_read_window < end_pos ?
                                 m_read_window : end_pos - cur_pos));
      cur_pos += m_read_window;
      m_read_window= new_window;
    }

    m_prc->message_received(m_msg_size);

  }
  catch(...)
  {
    save_error();
    return;
  }

  if (m_skip)
    return;

  // Parse message.

  scoped_ptr<Message> m_msg;
  m_msg.reset(mk_message(m_proto.m_side, m_msg_type));

  if (m_msg_size > 0)
  {
    try {
      assert(m_msg_size < (size_t)std::numeric_limits<int>::max());
      if (!m_msg->ParseFromArray(m_proto.m_rd_buf, (int)m_msg_size))
        throw_error(cdkerrc::protobuf_error, "Message could not be parsed");
    }
    catch (...)
    {
      save_error();
      m_msg.reset();
      return;
    }
  }

#ifdef DEBUG_PROTOBUF

  using std::cerr;
  using std::endl;

  cerr << endl;
  cerr << "<<<< Received message <<<<" << endl;
  cerr << "of type " << m_msg_type <<": "
       << msg_type_name(SERVER, m_msg_type) << endl;
  cerr << m_msg->DebugString();
  cerr << "<<<<" << endl << endl;

#endif

  // Pass data from parsed message to processor

  process_msg(m_msg_type, *m_msg);
}


/*
  Finish processing the current message and optionally start reading
  the next one (if read_next is true). If no more messages are read and
  there is saved error, throw it.

  Returns false if the operation should process another message, true
  if it is completed now.
*/

bool Op_rcv::finish(bool read_next)
{
  if (read_next)
  {
    read_msg();
    return false;
  }

  m_completed= true;
  throw_saved_error();
  return true;
}

}}}  // cdk::protocol::mysqlx



/*
  Explicit specialization is needed because Protocol constructor is templated
  (and thus no code is generated unless template is used).
*/
template cdk::foundation::opaque_impl<Protocol>::opaque_impl(void*, Protocol::Stream*);
template cdk::foundation::opaque_impl<Protocol_server>::opaque_impl(void*, Protocol::Stream*);


/*
  Implementation of Protocol methods using the internal implementation.
*/

namespace cdk {
namespace protocol {
namespace mysqlx {


// Client-side API

class Rcv_reply : public Op_rcv
{
public:

  Rcv_reply(Protocol_impl &proto) : Op_rcv(proto)
  {}

  void resume(Reply_processor &prc)
  {
    read_msg(prc);
  }

  Next_msg do_next_msg(msg_type_t type)
  {
    return msg_type::Ok == type ? EXPECTED : UNEXPECTED;
  }

  void do_process_msg(msg_type_t type, Message &msg)
  {
    if (msg_type::Ok != type)
      THROW("wrong message type");

    Mysqlx::Ok &ok= static_cast<Mysqlx::Ok&>(msg);
    static_cast<Reply_processor&>(*m_prc).ok(ok.msg());
  };

};


Protocol::Op& Protocol::snd_Close()
{
  Mysqlx::Connection::Close close;
  return get_impl().snd_start(close, msg_type::cli_Close);
}

Protocol::Op& Protocol::rcv_Reply(Reply_processor &prc)
{
  return get_impl().rcv_start<Rcv_reply>(prc);
}


// Server-side API
// ===============
// TODO: Complete and adapt to protocol changes.


Protocol::Op& Protocol_server::snd_Ok(const string &msg)
{
  Mysqlx::Ok ok;
  ok.set_msg(msg);
  return get_impl().snd_start(ok, msg_type::Ok);
}

Protocol::Op& Protocol_server::snd_Error(short unsigned errc, const string &msg)
{
  // TODO: SQL state
  Mysqlx::Error err;
  err.set_severity(Mysqlx::Error_Severity_ERROR);
  err.set_sql_state("SQLST");
  err.set_code(errc);
  err.set_msg(msg);
  return get_impl().snd_start(err, msg_type::Error);
}


class Rcv_command : public Op_rcv
{
public:

  Rcv_command(Protocol_impl &proto) : Op_rcv(proto)
  {}

  void resume(Cmd_processor &prc)
  {
    read_msg(prc);
  }

  Next_msg next_msg(msg_type_t type)
  {
    switch (type)
    {
    case msg_type::cli_Close:
      return EXPECTED;
    default: return UNEXPECTED;
    }
  }

  void process_msg(msg_type_t, Message&);
};


void Rcv_command::process_msg(msg_type_t type, Message &msg)
{
  Cmd_processor prc= static_cast<Cmd_processor&>(*m_prc);

  switch (type)
  {
  case msg_type::cli_Close: prc.close(); return;
  default: THROW("not implemented command");
  }
};


Protocol::Op& Protocol_server::rcv_Command(Cmd_processor &prc)
{
  return get_impl().rcv_start<Rcv_command>(prc);
}


// ------------------------------------------------------------

size_t Processor_base::message_begin_internal(msg_type_t type, bool &flag)
{
  m_type = type;
  return message_begin(type, flag);
}

void Processor_base::message_received_internal(size_t bytes_read)
{
  m_bytes_read = bytes_read;
  message_received(bytes_read);
}

}}}
