test_peer.cc

来自「BCAST Implementation for NS2」· CC 代码 · 共 850 行 · 第 1/2 页

// -*- c-basic-offset: 4; tab-width: 8; indent-tabs-mode: t -*-

// Copyright (c) 2001-2003 International Computer Science Institute
//
// Permission is hereby granted, free of charge, to any person obtaining a
// copy of this software and associated documentation files (the "Software")
// to deal in the Software without restriction, subject to the conditions
// listed in the XORP LICENSE file. These conditions include: you must
// preserve this copyright notice, and you cannot mention the copyright
// holders in advertising related to the Software without their permission.
// The Software is provided WITHOUT ANY WARRANTY, EXPRESS OR IMPLIED. This
// notice is a summary of the XORP LICENSE file; the license in that file is
// legally binding.

#ident "$XORP: xorp/bgp/harness/test_peer.cc,v 1.19 2003/09/16 18:13:46 pavlin Exp $"

// #define DEBUG_LOGGING
#define DEBUG_PRINT_FUNCTION_NAME

#include "bgp/bgp_module.h"
#include "config.h"

#include "libxorp/debug.h"
#include "libxorp/xlog.h"
#include "libxorp/status_codes.h"

#include "libxorp/eventloop.hh"
#include "libxipc/xrl_std_router.hh"
#include "bgp/socket.hh"

#include "xrl/interfaces/datain_xif.hh"
#include "bgppp.hh"
#include "test_peer.hh"

static const char SERVER[] = "test_peer";/* This servers name */
static const char SERVER_VERSION[] = "0.1";

#define	ZAP(s)	zap(s, #s)


/*
-------------------- XRL --------------------
*/

XrlTestPeerTarget::XrlTestPeerTarget(XrlRouter *r, TestPeer& test_peer,
				     bool trace)
    : XrlTestPeerTargetBase(r), _test_peer(test_peer), _trace(trace)
{
    debug_msg("\n");
}

XrlCmdError
XrlTestPeerTarget::common_0_1_get_target_name(string& name)
{
    debug_msg("\n");

    name = SERVER;
    return XrlCmdError::OKAY();
}

XrlCmdError
XrlTestPeerTarget::common_0_1_get_version(string& version)
{
    debug_msg("\n");

    version = SERVER_VERSION;
    return XrlCmdError::OKAY();
}

XrlCmdError
XrlTestPeerTarget::common_0_1_get_status(// Output values,
					 uint32_t& status,
					 string& reason)
{
    //XXX placeholder only
    status = PROC_READY;
    reason = "Ready";
    return XrlCmdError::OKAY();
}

XrlCmdError
XrlTestPeerTarget::common_0_1_shutdown()
{
    string error_string;
    if(!_test_peer.terminate(error_string)) {
	return XrlCmdError::COMMAND_FAILED(error_string);
    }

    return XrlCmdError::OKAY();
}


XrlCmdError
XrlTestPeerTarget::test_peer_0_1_register(const string& coordinator,
					  const uint32_t& genid)
{
    debug_msg("\n");

    if(_trace)
	printf("register(%s,%u)\n", coordinator.c_str(), genid);

    _test_peer.register_coordinator(coordinator);
    _test_peer.register_genid(genid);

    return XrlCmdError::OKAY();
}

XrlCmdError
XrlTestPeerTarget::test_peer_0_1_packetisation(const string& protocol)
{
    debug_msg("\n");

    if(_trace)
	printf("packetisation(%s)\n", protocol.c_str());

    if(!_test_peer.packetisation(protocol))
	return XrlCmdError::COMMAND_FAILED(c_format("Unsupported protocol %s",
					     protocol.c_str()));

    return XrlCmdError::OKAY();
}

XrlCmdError
XrlTestPeerTarget::test_peer_0_1_connect(const string&	host,
					 const uint32_t&  port)
{
    debug_msg("\n");

    if(_trace)
	printf("connect(%s,%u)\n", host.c_str(), port);

    string error_string;
    if(!_test_peer.connect(host, port, error_string)) {
	return XrlCmdError::COMMAND_FAILED(error_string);
    }

    return XrlCmdError::OKAY();
}

XrlCmdError
XrlTestPeerTarget::test_peer_0_1_listen(const string& address,
					const uint32_t&	port)
{
    debug_msg("\n");

    if(_trace)
	printf("listen(%s,%u)\n", address.c_str(), port);

    string error_string;
    if(!_test_peer.listen(address, port, error_string)) {
	return XrlCmdError::COMMAND_FAILED(error_string);
    }

    return XrlCmdError::OKAY();
}

XrlCmdError
XrlTestPeerTarget::test_peer_0_1_send(const vector<uint8_t>& data)
{
    debug_msg("\n");

    if(_trace && MESSAGETYPEOPEN == data[BGP_COMMON_HEADER_LEN - 1]) {
	printf("send() - open\n");
    }

    string error_string;
    if(!_test_peer.send(data, error_string)) {
	return XrlCmdError::COMMAND_FAILED(error_string);
    }

    return XrlCmdError::OKAY();
}

XrlCmdError
XrlTestPeerTarget::test_peer_0_1_disconnect()
{
    debug_msg("\n");

    if(_trace)
	printf("disconnect()\n");

    string error_string;
    if(!_test_peer.disconnect(error_string)) {
	return XrlCmdError::COMMAND_FAILED(error_string);
    }

    return XrlCmdError::OKAY();
}

XrlCmdError
XrlTestPeerTarget::test_peer_0_1_reset()
{
    debug_msg("\n");

    if(_trace)
	printf("reset()\n");

    _test_peer.reset();

    return XrlCmdError::OKAY();
}

XrlCmdError
XrlTestPeerTarget::test_peer_0_1_terminate()
{
    debug_msg("\n");

    if(_trace)
	printf("terminate()\n");

    string error_string;
    if(!_test_peer.terminate(error_string)) {
	return XrlCmdError::COMMAND_FAILED(error_string);
    }

    return XrlCmdError::OKAY();
}

/*
-------------------- IMPLEMENTATION --------------------
*/
TestPeer::TestPeer(EventLoop& eventloop, XrlRouter& xrlrouter,
		   const char *server, bool verbose)
    : _eventloop(eventloop), _xrlrouter(xrlrouter), _server(server),
      _verbose(verbose),
     _done(false),
      _s(UNCONNECTED), _async_writer(0), _listen(UNCONNECTED),
      _bgp(false),
      _flying(0),
     _bgp_bytes(0)
{
}

TestPeer::~TestPeer()
{
    delete _async_writer;
}

bool
TestPeer::done()
{
    return _done;
}

void
TestPeer::register_coordinator(const string& name)
{
    debug_msg("Test Peer: coordinator name = \"%s\"\n", name.c_str());

    _coordinator = name;
}

void
TestPeer::register_genid(const uint32_t& genid)
{
    debug_msg("Test Peer: genid = %u\n", genid);

    _genid = genid;
}

bool
TestPeer::packetisation(const string& protocol)
{
    debug_msg("\n");

    /*
    ** We only support "bgp".
    */
    if(protocol != "bgp") {
	return false;
    }

    _bgp = true;
    _bgp_bytes = 0;
    return true;
}

bool
TestPeer::connect(const string& host, const uint32_t& port,
		  string& error_string)
{
    debug_msg("\n");

    if(UNCONNECTED != _listen) {
	error_string = "Peer is currently listening";
	return false;
    }

    if(UNCONNECTED != _s) {
	error_string = "Peer is already connected";
	return false;
    }

    int s = ::socket(PF_INET, SOCK_STREAM, 0);
    if(-1 == s) {
	error_string = c_format("socket call failed: %s", strerror(errno));
	return false;
    }

    struct sockaddr_in peer;
    try {
	Socket::init_sockaddr(&peer,
			      Iptuple::get_addr(host.c_str()),
			      htons(port));
    } catch(UnresolvableHost e) {
	::close(s);
	error_string = e.why();
	return false;
    }

    if(-1 == ::connect(s, reinterpret_cast<struct sockaddr *>(&peer),
		       sizeof(peer))) {
	::close(s);
	return false;
    }

   /*
   ** If there is a coordinator then add a selector.
   */
   if(0 != _coordinator.length() &&
       !_eventloop.add_selector(s, SEL_RD,
				callback(this, &TestPeer::receive))) {
	::close(s);
	XLOG_WARNING("Failed to add socket %d to eventloop", s);
	return false;
    }

   /*
   ** Set up the async writer.
   */
    if(fcntl(s, F_SETFL, O_NONBLOCK) < 0) {
        XLOG_FATAL("Failed to go non-blocking: %s", strerror(errno));
    }
    delete _async_writer;
   _async_writer = new AsyncFileWriter(_eventloop, s);

    _s = s;

    return true;
}

bool
TestPeer::listen(const string& host, const uint32_t& port,
		 string& error_string)
{
    debug_msg("\n");

    if(UNCONNECTED != _listen) {
	error_string = "Peer is already listening";
	return false;
    }

    if(UNCONNECTED != _s) {
	error_string = "Peer is already connected";
	return false;
    }

    int s = ::socket(PF_INET, SOCK_STREAM, 0);
    if(-1 == s) {
	error_string = c_format("socket call failed: %s", strerror(errno));
	return false;
    }

    struct sockaddr_in local;
    try {
	Socket::init_sockaddr(&local,
			      Iptuple::get_addr(host.c_str()),
			      htons(port));
    } catch(UnresolvableHost e) {
	::close(s);
	error_string = e.why();
	return false;
    }

    int opt = 1;
    if(-1 == ::setsockopt(s, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt))) {
	error_string = c_format("setsockopt failed: %s\n", strerror(errno));
	return false;
    }

    if(-1 == ::bind(s, reinterpret_cast<struct sockaddr *>(&local),
		  sizeof(local))) {
	::close(s);
	error_string = c_format("Bind failed: %s", strerror(errno));
	return false;
    }

    if(-1 == ::listen(s, 1)) {
	::close(s);
	error_string = c_format("Bind failed: %s", strerror(errno));
	return false;
    }

    if(!_eventloop.add_selector(s, SEL_RD,
				callback(this,
					 &TestPeer::connect_attempt))) {
	::close(s);
	error_string = c_format("Failed to add socket %d to eventloop", s);
	return false;
    }
    _listen = s;

    return true;
}

bool
TestPeer::send(const vector<uint8_t>& data, string& error_string)
{
    debug_msg("len: %u\n", (uint32_t)data.size());
    if(UNCONNECTED == _s) {
	XLOG_WARNING("Not connected");
	error_string = "Not connected";
	return false;
    }

    size_t len = data.size();
    uint8_t *buf = new uint8_t[len];
    size_t i;
    for(i = 0; i < len; i++)
	buf[i] = data[i];
    _async_writer->add_buffer(buf, len,
			      callback(this, &TestPeer::send_complete));
    _async_writer->start();

    if(_verbose && _bgp)