Scale Stack 0.3 Developer Documentation

scalestack/database_proxy/client.cc

Go to the documentation of this file.

/*
 * Scale Stack
 *
 * Copyright 2010 Eric Day
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include "config.h"

#include <scalestack/database_proxy/client.h>
#include <scalestack/database_proxy/client_service.h>
#include <scalestack/database_proxy/connection.h>
#include <scalestack/kernel/module.h>

namespace scalestack
{
namespace database_proxy
{

/*
 * Public methods.
 */

client::client(client_service& client_service):
  network::stream(client_service),
  _state(),
  _next_state(),
  _need_yield(false),
  _shutdown(false),
  _size(),
  _packet_size(),
  _buffer(),
  _connection()
{
}

void client::set_connection(database_proxy::connection* bound_connection,
                            bool need_handshake)
{
  _connection = bound_connection;

  if (_connection == NULL)
    return;

  if (need_handshake && _state != SERVER_HANDSHAKE_PACKET &&
      _state != SERVER_HANDSHAKE_FLUSH)
  {
    _state = SERVER_HANDSHAKE_PACKET;
    reconnect();
  }
  else
    run_now();
}

void client::yield(void)
{
  _need_yield = true;
}

void client::connected(void)
{
  if (_connection == NULL)
    return;

  _connection->continue_write();
}

size_t client::read(uint8_t* buffer, size_t size)
{
  _size = size;
  _buffer = buffer;
  return _connection_write();
}

void client::continue_write(void)
{
  consume(_connection_write());
}

void client::flush_write(void)
{
  _connection->continue_write();
}

void client::error(void)
{
  _connection->error();
}

void client::run(void)
{
  if (_shutdown)
  {
    _connection->run_now();
    stop();
  }
  else
    _connection->continue_write();
}

/*
 * Private methods.
 */

size_t client::_connection_write(void)
{
  if (_connection == NULL)
    return 0;

  size_t written = 0;
  size_t total_written = 0;

  switch (_state)
  {
  case SERVER_HANDSHAKE_PACKET:
    if (_size < 4)
      return 0;

    _packet_size = 4 + (_buffer[0] | (static_cast<size_t>(_buffer[1]) << 8) | (static_cast<size_t>(_buffer[2]) << 16));
    _state = SERVER_HANDSHAKE_FLUSH;
    _module.log_debug("S-HAND packet: %u", _packet_size);

  case SERVER_HANDSHAKE_FLUSH:
    total_written = _connection->write(_buffer, _packet_size, true);
    _size -= total_written;
    _buffer += total_written;

    _packet_size -= total_written;
    if (_packet_size == 0)
      _state = RESULT_PACKET;

    break;

  case RESULT_PACKET:
    if (_size < 5)
      return 0;

    _packet_size = 4 + (_buffer[0] | (static_cast<size_t>(_buffer[1]) << 8) | (static_cast<size_t>(_buffer[2]) << 16));
    _state = RESULT_FLUSH;
    _module.log_debug("RESULT packet: %u", _packet_size);

    if (_buffer[4] == 0 || _buffer[4] >= 254)
      _next_state = RESULT_PACKET;
    else
      _next_state = COLUMN_PACKET;

  case RESULT_FLUSH:
    total_written = _connection->write(_buffer, _packet_size,
                                       _next_state == RESULT_PACKET);
    _size -= total_written;
    _buffer += total_written;

    _packet_size -= total_written;
    if (_packet_size == 0)
      _state = _next_state;

    if (_state != COLUMN_PACKET)
      break;

  case COLUMN_PACKET:
    while (1)
    {
      if (_size < 5)
        return 0;

      _packet_size = 4 + (_buffer[0] | (static_cast<size_t>(_buffer[1]) << 8) | (static_cast<size_t>(_buffer[2]) << 16));
      _state = COLUMN_FLUSH;
      _module.log_debug("COLUMN packet: %u", _packet_size);

      if (_packet_size == 9 && _buffer[4] == 254)
        _next_state = ROW_PACKET;
      else
        _next_state = COLUMN_PACKET;

  case COLUMN_FLUSH:
      written = _connection->write(_buffer, _packet_size, false);
      _size -= written;
      _buffer += written;
      total_written += written;

      _packet_size -= written;
      if (_packet_size == 0)
        _state = _next_state;

      if (_state == ROW_PACKET)
        break;
    }

  case ROW_PACKET:
    while (1)
    {
      if (_size < 5)
        return 0;

      _packet_size = 4 + (_buffer[0] | (static_cast<size_t>(_buffer[1]) << 8) | (static_cast<size_t>(_buffer[2]) << 16));
      _state = ROW_FLUSH;
      _module.log_debug("ROW packet: %u", _packet_size);

      if (_packet_size == 9 && _buffer[4] == 254)
        _next_state = RESULT_PACKET;
      else
        _next_state = ROW_PACKET;

  case ROW_FLUSH:
      written = _connection->write(_buffer, _packet_size,
                                   _next_state == RESULT_PACKET);
      _size -= written;
      _buffer += written;
      total_written += written;

      _packet_size -= written;
      if (_packet_size == 0)
        _state = _next_state;

      if (_state == RESULT_PACKET)
        break;
    }
  }

  if (_need_yield && _state == RESULT_PACKET && _packet_size == 0)
  {
    _need_yield = false;
    _connection->yield();
  }

  return total_written;
}

} /* namespace database_proxy */
} /* namespace scalestack */

---