// -*- c++ -*-
// Generated by assa-genesis
//------------------------------------------------------------------------------
// $Id: sb_tests.cpp,v 1.7 2006/07/20 02:30:56 vlg Exp $
//------------------------------------------------------------------------------
//                            sb_tests.cpp
//------------------------------------------------------------------------------
//  Copyright (c) 2005 by Vladislav Grinchenko 
//
//  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; either version  
//  2 of the License, or (at your option) any later version.      
//------------------------------------------------------------------------------
//
// Date   : Sun Jun 26 12:10:13 2005
//
//------------------------------------------------------------------------------

static const char help_msg[]=
"                                                                            \n"
" NAME:                                                                      \n"
"                                                                            \n"
"   sb_tests                                                                 \n"
"                                                                            \n"
" DESCRIPTION:                                                               \n"
"                                                                            \n"
"   sb_tests is the server-side of the SocketBuffer overflow test.           \n"
"   See the description of sb_testc for all details.                         \n"
"                                                                            \n"
" USAGE:                                                                     \n"
"                                                                            \n"
"   shell>  sb_tests --data-port=NAME --ctrl-port=NAME                       \n"
"                    --output-file=NAME [OPTIONS]                            \n"
"                                                                            \n"
" OPTIONS:                                                                   \n"
"                                                                            \n"
"     --data-port NAME    - Listening data port (PORT[@HOST])                \n"
"     --ctrl-port NAME    - Data flow control port of the client program.    \n"
"     --output-file NAME  - File to transfer to the server side.             \n"
"                           (default: sbtest-data.out)                       \n"
"                                                                            \n"
"     --build-dir PATH    - Directory where executables were built.          \n"
"                                                                            \n"
" -b, --daemon            - Run process as true UNIX daemon                  \n"
" -l, --pidfile PATH      - The process ID is written to the lockfile PATH   \n"
"                           instead of default ~/.{procname}.pid             \n"
" -L, --ommit-pidfile     - Do not create PID lockfile                       \n"
"                                                                            \n"
" -D, --log-file NAME     - Write debug to NAME file                         \n"
" -d, --log-stdout        - Write debug to standard output                   \n"
" -z, --log-size NUM      - Maximum size debug file can reach (dfl: is 10Mb) \n"
"                                                                            \n"
" -c, --log-level NUM     - Log verbosity                                    \n"
" -s, --with-log-server   - Redirect log messages to the log server          \n"
" -S, --log-server NAME   - Define assa-logd server address                  \n"
"                                                                            \n"
" -m, --mask MASK         - Mask (default: ALL = 0x7fffffff)                 \n"
"                                                                            \n"
" -h, --help              - Print this messag                                \n"
" -v, --version           - Print version number                            \n";
//------------------------------------------------------------------------------

#ifdef HAVE_CONFIG_H
#    include "config.h"
#endif
#include <string>
using std::string;

#include <fstream>

#include <assa/GenServer.h>
#include <assa/Singleton.h>
#include <assa/TimeVal.h>
#include <assa/Assure.h>
#include <assa/IPv4Socket.h>
#include <assa/ServiceHandler.h>
#include <assa/CharInBuffer.h>
#include <assa/Connector.h>
#include <assa/Acceptor.h>
#include <assa/INETAddress.h>

using namespace ASSA;

enum { SBTS   = ASSA::APP,
	   SBDATA = ASSA::USR1		// HEX/ASCII incoming data packets dump
};

class DataReader;

//------------------------------------------------------------------------------
// FlowCtrl class
//------------------------------------------------------------------------------
class FlowCtrl : public ServiceHandler<IPv4Socket>
{
public:
	FlowCtrl ();

	virtual int open ();
	virtual int handle_read   (int fd_);
	virtual int handle_close  (int fd_);

	void emmit_drained_signal ();

private:
	CharInBuffer m_input;
	std::string  m_drained_msg;	// OUT message
	std::string  m_ready_msg;	// OUT message
	std::string  m_blocked_msg;	// IN  message
};

//------------------------------------------------------------------------------
// Xmit Byte counter
//------------------------------------------------------------------------------
class XmitByteCounter 
{
public:
	XmitByteCounter () : m_count (0) { }
	~XmitByteCounter () {
		DL ((ASSA::APP,"Total bytes processed %d\n", m_count));
	}
	void add (size_t count_) { m_count += count_; }

private:
	size_t m_count;
};

//------------------------------------------------------------------------------
// DataReader class
//------------------------------------------------------------------------------
class DataReader : public ServiceHandler<IPv4Socket>
{
public:
	DataReader (IPv4Socket* stream_);

	virtual int open ();
	int drain_pipe ();

private:
	Connector<FlowCtrl, IPv4Socket> m_connector;

	FlowCtrl       m_flow_ctrl;
	std::ofstream  m_sink;
	size_t         m_total_rcvd; // total bytes received
};

//------------------------------------------------------------------------------
// SocketBuffer Test Server
//------------------------------------------------------------------------------
class SBTestServer :
    public ASSA::GenServer,
    public ASSA::Singleton<SBTestServer>
{
public:
    SBTestServer ();

    virtual void init_service    ();
    virtual void process_events ();

	std::string get_build_dir   () const { return m_build_dir;   }
	std::string get_data_port   () const { return m_data_port;   }
	std::string get_ctrl_port   () const { return m_ctrl_port;   }
	std::string get_output_file () const { return m_output_file; }

	DataReader* get_data_reader () { return m_data_reader; }
	void register_reader (DataReader* reader_) { m_data_reader = reader_; }
	
	void abort_test () { stop_service (); }

private:
	std::string m_build_dir;
	std::string m_data_port;
	std::string m_ctrl_port;
	std::string m_output_file;

	DataReader* m_data_reader;
	Acceptor<DataReader, IPv4Socket>* m_acceptor;
};

/* Useful definitions */

#define SBTESTSERVER  SBTestServer::get_instance()
#define REACTOR       SBTESTSERVER->get_reactor()
#define DATAREADER    SBTESTSERVER->get_data_reader()

// Static declarations mandated by Singleton class
ASSA_DECL_SINGLETON(SBTestServer);

//------------------------------------------------------------------------------
// FlowCtrl member functions
//------------------------------------------------------------------------------
FlowCtrl::
FlowCtrl ()
	: 
	m_input (32, "\n"), 
	m_drained_msg ("drained\n"), // outgoing msg
	m_ready_msg   ("ready\n"),	 // outgoing msg
	m_blocked_msg ("blocked")	 // incoming msg
{
	trace ("FlowCtrl::");
}

int
FlowCtrl::
open ()
{
	trace ("FlowCtrl::open");

    ASSA::IPv4Socket& s = *this;
    REACTOR->registerIOHandler (this, s.getHandler (), ASSA::READ_EVENT);

	/* emmit 'ready' signal 
	 */
	s.write (m_ready_msg.c_str (), m_ready_msg.length ());
	s << ASSA::flush;

	return 0;
}

/**
 * The only message we expect from sb_testc client is 'blocked[CR]'
 */
int
FlowCtrl::
handle_read (int fd_)
{
	trace ("FlowCtrl::handle_read");

    ASSA::IPv4Socket& s = *this;
    if (s.getHandler () != fd_) { return (-1); }

	s >> m_input;

	if (m_input) {				
		if (m_input.c_str () == m_blocked_msg) {
			DL ((SBTS, "=> received 'blocked' signal from test client.\n"));
			if (DATAREADER->drain_pipe () < 0) {
				return -1;
			}
			m_input.reset ();
			emmit_drained_signal ();
		}
	}
	else {
		if (m_input.state () == ASSA::CharInBuffer::error) {
			DL ((SBTS,"Input buffer overflow!\n"));
			return -1;
		}
	}

	return BYTES_LEFT_IN_SOCKBUF(s);
}

int
FlowCtrl::
handle_close (int fd_)
{
	trace ("FlowCtrl::handle_close");

	get_stream ().close ();
	DATAREADER->drain_pipe ();
	SBTESTSERVER->abort_test ();

	return 0;
}

void
FlowCtrl::
emmit_drained_signal ()
{
	trace ("FlowCtrl::emmit_drained_signal");

	get_stream ().write (m_drained_msg.c_str (), m_drained_msg.length ());
	get_stream () << ASSA::flush;
}

//------------------------------------------------------------------------------
// DataReader member functions
//------------------------------------------------------------------------------
DataReader::
DataReader (IPv4Socket* stream_) : 
	ServiceHandler<IPv4Socket> (stream_),
	m_total_rcvd (0)
{
	trace ("DataReader::DataReader");
	const char* fname = SBTESTSERVER->get_output_file ().c_str ();

	::unlink (fname);

	m_sink.open (fname);
	if (!m_sink) {
		DL ((SBTS,"Failed to open output data file.\n"));
		SBTESTSERVER->abort_test ();
	}
}

int
DataReader::
open ()
{
	trace ("DataReader::open");
	int ret;

	DL ((SBTS,"=> New data client connected.\n"));
	SBTESTSERVER->register_reader (this);

	INETAddress ctrl_addr (SBTESTSERVER->get_ctrl_port ().c_str ());
	Assure_exit (!ctrl_addr.bad ());

	m_connector.open ();
	ret = m_connector.connect (&m_flow_ctrl, ctrl_addr);

	if (ret < 0) {
		DL ((SBTS,"Failed to connect to 'sb_testc' (%s:%d)\n",
			 ctrl_addr.getHostName ().c_str (), ctrl_addr.getPort ()));
		SBTESTSERVER->abort_test ();
	}
	else {
		DL ((SBTS,"Connected to CTRL port of 'sb_testc'\n"));
	}

	return 0;
}

int
DataReader::
drain_pipe ()
{
	trace ("DataReader::drain_pipe");

	const size_t bufsz = Streambuf::MAXTCPFRAMESZ;
	char buf [bufsz];
	int len = 0;
	XmitByteCounter total;

	while (get_stream () && (len = get_stream ().read (buf, bufsz)) > 0) 
	{
		m_sink.write (buf, len);     // could be binary data

		m_total_rcvd += len;
		total.add (len);

		DL ((SBTS,"=> got %d bytes (out of %d intended)\n", 
			 len, Streambuf::MAXTCPFRAMESZ));
		MemDump::dump_to_log (SBDATA,"=> received data from client", buf, len);
	}

	get_stream ().dumpState ();
	DL ((SBTS,"get_stream().read() = %d\n", len));

	if (!get_stream ()) {
		DL ((SBTS,"=> Client side closed connection\n"));
		get_stream ().close ();
		m_sink << std::flush;
		m_sink.close ();
		DL ((SBTS, "Received grand total %d bytes\n", m_total_rcvd)); 
		return -1;
	}
	return 0;
}


//------------------------------------------------------------------------------
// SBTestServer member functions
//------------------------------------------------------------------------------
SBTestServer::
SBTestServer () : 
	m_acceptor (NULL),
	m_output_file ("sbtest-data.out")
{
	add_opt (0, "build-dir",   &m_build_dir);
	add_opt (0, "data-port",   &m_data_port);
	add_opt (0, "ctrl-port",   &m_ctrl_port);
	add_opt (0, "output-file", &m_output_file);

    // ---Configuration---
    rm_opt ('f', "config-file"  );
    rm_opt ('n', "instance"     );

    /*---
     * By defauil disable all debugging
     *---*/
    m_mask = SBTS | ASSA::ASSAERR | ASSA::TRACE;
    m_log_file = "sbtest-server.log";
}


void
SBTestServer::
init_service ()
{
    trace("SBTestServer::init_service");

	if (m_build_dir.length () == 0) {
		DL ((SBTS,"Missing {--build-dir=PATH} option!\n"));
		abort_test ();
		return;
	}

	if (m_data_port.length () == 0) {
		DL ((SBTS,"Missing {--data-port=STRING} option!\n"));
		abort_test ();
		return;
	}

	if (m_ctrl_port.length () == 0) {
		DL ((SBTS,"Missing {--ctrl-port=STRING} option!\n"));
		abort_test ();
		return;
	}

	if (m_output_file.length () == 0) {
		DL ((SBTS,"Missing {--output-file=STRING} option!\n"));
		abort_test ();
		return;
	}

	/** Initialize Acceptor
	 */
	m_acceptor = new Acceptor<DataReader, IPv4Socket> (REACTOR);

	/** Open listening socket
	 */
	INETAddress listen_addr (get_data_port ().c_str ());
	Assure_exit (!listen_addr.bad ());
	Assure_exit (m_acceptor->open (listen_addr) == 0);

	DL ((SBTS,"Listening socket opened on %s:%d\n", 
		 listen_addr.getHostName ().c_str (),
		 listen_addr.getPort ()));

    DL((SBTS,"Service has been initialized.\n"));
}

void
SBTestServer::
process_events ()
{
    trace("SBTestServer::process_events");

    while (service_is_active ()) {
        m_reactor.waitForEvents ();
    }

    // Shut the service down
    m_reactor.stopReactor ();
    DL((ASSA::APP,"Service stopped!\n"));
}



int
main (int argc, char* argv[])
{
    static const char release[] = "VERSION";
    int patch_level = 0;

    SBTESTSERVER->set_version (release, patch_level);
    SBTESTSERVER->set_author  ("Vladislav Grinchenko");
    SBTESTSERVER->set_flags   (ASSA::GenServer::RMLOG);

    SBTESTSERVER->init (&argc, argv, help_msg);
 
    SBTESTSERVER->init_service ();
    SBTESTSERVER->process_events ();

    return SBTESTSERVER->get_exit_value ();
}