File Coverage

blib/lib/Net/PSYC/Circuit.pm

Criterion	Covered	Total	%
statement	256	348	73.5
branch	95	170	55.8
condition	40	81	49.3
subroutine	33	44	75.0
pod	0	3	0.0
total	424	646	65.6

package Net::PSYC::Circuit;

our $VERSION = '0.4';

use strict;
use Socket qw(SO_KEEPALIVE inet_ntoa);
use IO::Socket::INET;

import Net::PSYC qw( watch add W sendmsg same_host send_mmp parse_uniform BLOCKING makeMSG make_psyc parse_psyc parse_mmp PSYC_PORT PSYCS_PORT register_host register_route make_mmp UNL);

sub listen {
	# looks funky.. eh? undef makes IO::Socket handle INADDR_ANY properly
	# whereas '' causes an exception. stupid IO::Socket if you ask me.
    my ($class, $ip, $port, $options) = @_;
    my $socket = IO::Socket::INET->new(
			LocalAddr => $ip || undef,
				    # undef == use INADDR_ANY
                        LocalPort => $port || undef,
				    # undef == take any port
                        Proto => 'tcp',
                        Listen => 7,
		        Blocking => BLOCKING() & 2,
			Timeout => 5,
                        ReuseAddr => 1
		    )
			or return $!;
    my $self = {
	'SOCKET' => $socket,
	'IP' => $ip||$socket->sockhost(),
	'PORT' => $_[2] || $socket->sockport,
	'LAST_RECV' => getsockname($socket),
	'type' => 'c',
	'O' => $options,
    };
    W1('TCP Listen %s:%s successful.', $self->{'IP'}, $self->{'PORT'});
    bless $self, 'Net::PSYC::Circuit::L';
    watch($self) unless BLOCKING() & 2;
    return $self;
}

#   new ( \*socket, vars )
sub new {
    my ($class, $socket, $vars) = @_;
    my $self = {
	'O' => {},
	'SOCKET' => $socket,
	'type' => 'c',
	# These buffer may be moved to a reconnected object.
	# maybe not the IN-buffer, but right now we cannot do anything
	# about it. -> later TODO
	'I_BUFFER' => '',
	'O_BUFFER' => [],
	# this one not! (its only used for negotiation and such)
	'N_BUFFER' => [],
	'O_COUNT' => 0,
	'CACHE' => {}, # cache for fragmented data
	'I_LENGTH' => 0, # whether _length of incomplete
			 # packets exceeds buffer-length
	'FRAGMENT_COUNT' => 0,
	'R' => {},
	'L' => 0,
	'state_temp' => {},
	'state' => {},
	'vars' => {},
	'error' => 0,
	%$vars,
    };
    $socket->sockopt( SO_KEEPALIVE(), 1 );
    bless $self, 'Net::PSYC::Circuit::C';

    $self->{'R_HOST'} =	$self->{'R_IP'};
    $self->{'peeraddr'} = "psyc://$self->{'R_HOST'}:$self->{'R_PORT'}/";
    $Net::PSYC::C{"$self->{'R_IP'}\:$self->{'R_PORT'}"} = $self;

    # stupid if.
    register_host($self->{'R_IP'}, $self->{'R_HOST'}) if ($self->{'R_HOST'});
    register_host('127.0.0.1', 'localhost');
    # TRUST ist something arbitrary anyway.. there is no problem to wait for
    # the dns_resolution to set it correctly
    same_host('127.0.0.1', $self->{'R_IP'},
	      sub {
		    my $result = shift;
		    $self->TRUST(9) if $result;
	      });
    register_route("$self->{'R_HOST'}\:$self->{'R_PORT'}", $self);

    # if we are an accepted socket, the greeting has allready been fired out
    #
    # in case we are blocking _and_ mmp modules have been activated we block
    # here and do negotiation
    #
    watch($self) unless (BLOCKING() & 2);
    if (BLOCKING() & 1) { # blocking writes!
	$self->logon();
    } else {
	add($self->{'SOCKET'}, 'w', sub { $self->logon() }, 0);
    }
    return $self;
}

sub connect {
    my $class = shift;
    my $ip = shift;
    my $port = shift || PSYC_PORT();
    my $socket = IO::Socket::INET->new(Proto     => 'tcp',
                                       PeerAddr  => $ip,
				       Blocking	=> BLOCKING() & 1,
                                       PeerPort  => $port );
    # we need some nonblocking error handling
    if (!$socket) {
	W1('TCP connect to %s:%d failed. (%s)', $ip, $port, $!);
	return 0;
    }
    my $self = {
	'R_IP' => $ip,
	'R_PORT' => $port,
    };
    return Net::PSYC::Circuit->new($socket, $self);
}

# TCP connection class
package Net::PSYC::Circuit::C;

use bytes;
use strict;

use Socket;

use base qw(Net::PSYC::MMP::State Net::PSYC::Hook);

import Net::PSYC qw( revoke W UNL sendmsg same_host send_mmp parse_uniform BLOCKING makeMSG make_psyc parse_psyc parse_mmp make_mmp register_route register_host dns_lookup);

my $PING_INTERVAL = 77;

sub TRUST {
    my $self = shift;
    $self->{'TRUST'} = $_[0] if exists $_[0];
    return $self->{'TRUST'} || 3;
}

sub accept_modules {
    my $self = shift;
    my $module = shift;
    my $on = shift;

    # !defined($on) ist quasi 1
    if (!defined($on) || $on) {
	return 1 if $self->accepting_modules($module);
	$self->{'O'}->{'_understand_modules'}->{$module} = 1;
	$self->fire($self->{'peeraddr'},0,{ '+_understand_modules' => $module });
	# it is possible that this happens before proper neg? then we are
	# dead! TODO

	return 1 if (!$self->{'R'}->{'_understand_modules'}->{$module});

	if ($module eq '_compress') {
	    return $self->zlib_init_client();
	} elsif ($module eq '_encrypt') {
	    unless (SSL()) {
		W0("The other side offers SSL-encryption. It would be wise to install IO::Socket::SSL (v0.93 or above).");
		return 1;
	    }

	    $self->fire($self->{'peeraddr'},0,
			{ '+_using_modules' => '_encrypt' },
			sub { $self->{'OK'} = 0 },
		    );
	    $self->{'SSL_client'} = 1;
	    # TODO . same code as in gotiate(). Think about something else.
	    # plus: in case we have eventing we should use a timer-event to
	    # stop waiting.
	    return 1;
	}
    } else {
	# may be impossible.
	W0('It is impossible to remove the mmp module %s from an established'.
	   ' connection.', $module);
	return 1 if (!exists $self->{'O'}->{'_understand_modules'}->{$module});
    }
}

sub accepting_modules {
    my $self = shift;
    my $module = shift;

    return 0 unless ($self->{'O'}->{'_understand_modules'}->{$module});
    return 1;
}

# counterparts to understand_ and use_
# means that a _using_modules came in
sub negotiate {
    my $self = shift;
    my $module = shift;

    unless (exists $self->{'R'}->{'_using_modules'}) {
	$self->{'R'}->{'_using_modules'} = {};
    }
    $self->{'R'}->{'_using_modules'}->{$module} = 1;

    if ($module eq '_encrypt') {
	if ($self->{'SSL_client'}) {
	    return $self->tls_init_client();
	} else {
	    return $self->tls_init_server();
	}
    } elsif ($module eq '_compress') {
	return $self->zlib_init_server();
    }
}

sub gotiate {
    my $self = shift;
    my $module = shift;

    unless (exists $self->{'R'}->{'_understand_modules'}) {
	$self->{'R'}->{'_understand_modules'} = {};
    }
    $self->{'R'}->{'_understand_modules'}->{$module} = 1;
    return 1 unless ($self->accepting_modules($module));

    if ($module eq '_encrypt') {
	unless (SSL()) {
	    W0("The other side offers SSL-encryption. It would be wise to install IO::Socket::SSL (v0.93 or above).");
	    return 1;
	}

if (Net::PSYC::FORK) {
	$self->fire($self->{'peeraddr'},0,
		    { '+_using_modules' => '_encrypt' },
		    sub { $self->{'OK'} = 0 }
		);
} else {
	$self->fire('',0,{ '_using_modules' => '_encrypt' },
		    sub { $self->{'OK'} = 0 }
		);
}
	$self->{'SSL_client'} = 1;

	return 1;
    } elsif ($module eq '_compress') {
	Net::PSYC::Event::revoke($self->{'SOCKET'}, 'w');
	$self->zlib_init_client();
	return 1;
    }
}

sub ping_init {
    my $self = shift;
    # we are a server or do not have eventing
    return 1 if ($self->{'L'} || BLOCKING());

    Net::PSYC::Event::remove($self->{'ping_id'}) if exists $self->{'ping_id'};
    $self->{'ping_sub'} ||= sub {
	syswrite($self->{'SOCKET'}, ".\n");
    };
    $self->{'ping_id'} = Net::PSYC::Event::add( $PING_INTERVAL, 't',
			      $self->{'ping_sub'}, 1);
}

sub tls_init_server { 1 }
sub tls_init_client {
    my $self = shift;
    my $t = IO::Socket::SSL->start_SSL($self->{'SOCKET'});
#	SSL_server => ($self->{'L'}) ? 1 : 0);
    if (ref $t ne 'IO::Socket::SSL') {
	return 1;
    }
    W1('Using encryption to %s.', $self->{'peeraddr'});
    $self->{'SOCKET'} = $t;
    $self->{'OK'} = 1;
    unless (BLOCKING()) {
	Net::PSYC::Event::forget($self);
	Net::PSYC::Event::watch($self);
	Net::PSYC::Event::revoke($self->{'SOCKET'}, 'w');
    }
}

# the naming of client/server is fucked up. TODO
sub zlib_init_server {
    my $self = shift;
    unless (eval{ require Net::PSYC::MMP::Compress }) {
	Net::PSYC::shutdown($self->{'SOCKET'});
	W0('Somehow your Compression modules does not work (%s). '.
	   'Shutting down connection.', $@);
	# shut down.. whatever
	# TODO switch off _understand_modules _compress
	return 1;
    }
    unless ($self->{'_compress'}) {
	$self->{'_compress'} = new Net::PSYC::MMP::Compress($self);
    }
    $self->{'_compress'}->init('decrypt');
    return 1;
}

sub zlib_init_client {
    my $self = shift;
    unless (eval { require Net::PSYC::MMP::Compress }) {
	W0('Somehow your Compression modules does not work (%s).', $@);
	return 1;
    }

    unless ($self->{'_compress'}) {
	$self->{'_compress'} = new Net::PSYC::MMP::Compress($self);
    }
if (Net::PSYC::FORK) {
    $self->fire($self->{'peeraddr'},0,{ '+_using_modules' => '_compress' },
		sub { $self->{'_compress'}->init('encrypt') });
} else {
    $self->fire('',0,{ '_using_modules' => '_compress' },
		sub { $self->{'_compress'}->init('encrypt') });
}
}

sub logon {
    my $self = shift;

    $self->{'O'} = \%Net::PSYC::O;
    # TODO nonblocking dns.
    $self->{'R_HOST'} = gethostbyaddr($self->{'SOCKET'}->peeraddr(), AF_INET())
			|| $self->{'R_IP'};
    $self->{'IP'} = $self->{'SOCKET'}->sockhost();
    $self->{'PORT'} = $self->{'SOCKET'}->sockport();
    $self->{'R_IP'} = $self->{'SOCKET'}->peerhost();
    $self->{'R_PORT'} = $self->{'SOCKET'}->peerport();
    $self->{'LAST_RECV'} = $self->{'SOCKET'}->peername();
    register_host($self->{'R_IP'}, inet_ntoa($self->{'SOCKET'}->peeraddr()));
    register_host('127.0.0.1', $self->{'IP'});
    register_route(inet_ntoa($self->{'SOCKET'}->peeraddr()).":$self->{'R_PORT'}", $self);

    W1('TCP: Connected with %s:%s', $self->{'R_IP'}, $self->{'R_PORT'});
    syswrite($self->{'SOCKET'}, ".\n");

    # I would like to rename OK. it may be necessary to work on STATE-BITS in
    # the future. for now this is okay TODO
    #
    # we allow sending messages before receiving a _notice_circuit_established
    # in case
    # 	- we accept()ed the connection
    # 	- we are doing blocking writes _and_ reads
    # 	- we are anachronistic and not on tls, zlib or lsd TODO
    unless (BLOCKING() & 1) {
	Net::PSYC::Event::add($self->{'SOCKET'}, 'w', sub {$self->write()}, 0);
    }
    if ($self->{'L'} || (BLOCKING() & 1 && BLOCKING() & 2)) {
	$self->{'OK'} = 1;
    }
    $self->greet();
}

# greet
sub greet {
    my $self = shift;

    # _notice_circuit_established versenden. (MMP-neg)
    # we _could_ send _using_modules here.. but. who cares???
    my $h;
    my $m;
if (Net::PSYC::FORK) {
    $h = {
	'=_understand_modules' => [ keys %{$self->{'O'}->{'_understand_modules'}} ],
	'=_implementation' => $self->{'O'}->{'_implementation'},
	'=_understand_protocols' => $self->{'O'}->{'_understand_protocols'},
    };
    $m = make_psyc('_notice_circuit_established',
		      'Connection to [_source] established!');
} else {
    $h = {
	'_understand_modules' => [ keys %{$self->{'O'}->{'_understand_modules'}} ],
	'_implementation' => $self->{'O'}->{'_implementation'},
	'_understand_protocols' => $self->{'O'}->{'_understand_protocols'},
    };
    $m = make_psyc('_notice_circuit_established',
		      'Connection to [_source] established!');
    $self->fire($self->{'peeraddr'}, $m, $h);
    # formally this is wrong, because in !FORK _*_modules are psyc vars. but
    # since the muve does not give a shit we do neither.
    $m = make_psyc('_status_circuit',
		      'I feel good.');

}
    $self->fire($self->{'peeraddr'}, $m, $h);
}

sub send {
    my ($self, $target, $data, $vars, $prio) = @_;

    W2('"%s" -> send(%.10s.., %s)', $self->{'peeraddr'}, $target, $data);
    if (!exists $vars->{"_source"} && exists $self->{'me'}) {
	$vars->{"_source"} = $self->{'me'};
    }
    if (ref $data eq 'ARRAY') {
	if (1) { #$self->{'O'}->{'_using_modules'}->{'_fragments'}) {
	    $vars->{'_counter'} = $self->{'FRAGMENT_COUNTER'}++;
	    $vars->{'_amount_fragments'} = scalar(@$data);
	} else {
	    # very bad bad idea... better drop the packet
	    $data = [ join('', @$data) ];
	}
    } else {
	$data = [ $data ];
    }

    push(@{$self->{'O_BUFFER'}}, [ $data, $vars, 0 ]);

    $self->{'O_COUNT'} = scalar(@{$self->{'O_BUFFER'}}) - 1 if ($prio);

    if (BLOCKING()) { # send the packet instantly
        $self->write();
    } elsif ($self->{'OK'}) {
	revoke($self->{'SOCKET'}, 'w');
    }

    return 0;
}

sub fire {
    my $self = shift;
    my ($target, $data, $vars, $cb) = @_;

    $data ||= '';
    $vars ||= {};
    $vars->{'_target'} = $target if $target;
    #unless ($vars->{'_target'}) {
    #	W("fire may not be called without a proper _target",0);
    #	return 0;
    #}
    $vars->{'_source'} ||= delete $vars->{'_source'};

    #W("'$vars->{'_target'}'->fire('$data', $vars, ".($cb||'undef').")",2);

    if (!exists $vars->{"_source"} && exists $self->{'me'}) {
	$vars->{"_source"} = $self->{'me'};
    }

    push(@{$self->{'N_BUFFER'}}, [ [ $data ], $vars, 0, $cb ]);
    if (BLOCKING()) { # send the packet instantly
        $self->write();
    } else {
	Net::PSYC::Event::revoke($self->{'SOCKET'}, 'w');
    }
}

sub write () {
    my $self = shift;

    # no permission to send packets.. and we are not wierdo enough!
    # TODO
    return 1 unless ($self->{'OK'});

    my $N = $self->{'N_BUFFER'};
    my $O = $self->{'O_BUFFER'};
    my ($data, $vars, $count, $cb);

    if (scalar @$N) {
	($data, $vars, $count, $cb) = @{$N->[0]};
    } elsif (exists $O->[$self->{'O_COUNT'}]) {
	($data, $vars, $count) = @{$O->[$self->{'O_COUNT'}]};
    } else {
	W2('packets in %p: %d%s', $self, scalar(@$O), "\n");
	return 1; # no packets!
    }

    $vars->{'_fragment'} = $count if ($vars->{'_amount_fragments'});

    my $d = $data->[$count];

    use Storable qw(dclone);
    $vars = dclone($vars); # but the current design.. TODO
    # TODO . shutdown connection if trigger fails. its really important for
    # encryption/decryption
    $self->trigger('send', $vars, \$d);

    my $m = make_mmp($vars, $d, $self);
    $self->trigger('encrypt', \$m);

    if (!defined(syswrite($self->{'SOCKET'}, $m))) {
	# put the packet back into the queue

	if (++$self->{'error'} >= 3) {
	    W0('Sending a tcp packet to %s failed for the third time. Closing '.
	       'connection.', $self->{'peeraddr'});
	    return -1;
	}
	W0('Sending a packet to %s failed (%s). %d more retries.',
	    $self->{'peeraddr'}, $self->{'error'});
	return 1;
    } else {
	$self->{'error'} = 0;
    }
    $self->ping_init();

    $self->trigger('sent', $vars, \$d);
    $cb->() if ($cb);

    W2('TCP: wrote %d bytes of data to the socket', length($m));
    W2('TCP: >>>>>>>> OUTGOING >>>>>>>>\n%s\nTCP: <<<<<<< OUTGOING <<<<<<<\n',
       $m);

    if (($vars->{'_amount_fragments'} || @$data) == $count + 1) {
	# all fragments of this packet sent
	# delete it..
	if (scalar @$N) {
	    shift @$N;
	} else {
	    splice(@{$O}, $self->{'O_COUNT'}, 1);
	}
    } else {
	# fragments of this packet left
	# increase the fragment-id
	$self->{'O_BUFFER'}->[$self->{'O_COUNT'}]->[2]++;
	# increase the packet id..
	$self->{'O_COUNT'}++;
    }
    $self->{'O_COUNT'} = 0 unless ( exists $O->[$self->{'O_COUNT'}] );
    if ( @$N || @$O ) {
	if (BLOCKING() || $Net::PSYC::ANACHRONISM) { # send the packet
	    $self->write();
	} else {
	    revoke($self->{'SOCKET'}, 'w');
	}
    }
    return 1;
}

sub read () {
    my $self = shift;
    my ($data, $read);

    # if you change the buffer-size.. remember to fix buffersize of
    # MMP::Compress and rest..
    $read = sysread($self->{'SOCKET'}, $data, 4096);

    return if (!$read); # connection lost !?
    # gibt es nen 'richtigen' weg herauszufinden, ob die connection noch lebt?
    # connected() und die ganzen anderen socket-funcs helfen einem da in
    # den ekligen fällen nicht..

    unless ($self->trigger('decrypt', \$data)) {
	W0('Fatal error during decrypt. Closing connection');
	return;
    }

    $$self{'I_BUFFER'} .= $data;
    warn $! unless (defined($read));
    $self->{'I_LENGTH'} += $read;
#    open(file, ">>$self->{'HOST'}:$self->{'PORT'}.in");
#    print file $data;
#    print file "\n========\n";
#    close file;
    W2('TCP: Read %d bytes from socket.', $read);
    W2('TCP: >>>>>>>> INCOMING >>>>>>>>\n%s\nTCP: <<<<<<< INCOMING <<<<<<<',
	$data);

    unless ($self->{'LF'}) {
	# we need to check for a leading ".\n"
	# this is not the very best solution though..
	if ($self->{'I_LENGTH'} > 2) {
	    if ( $self->{'I_BUFFER'} =~ s/^\.(\r?\n)//g ) {
		$self->{'LF'} = $1;
		# remember if the other side uses \n or \r\n
		# to terminate lines.. we need that for proper
		# and safe parsing
	    } else {
		syswrite($self->{'SOCKET'},
		    make_psyc('_error_syntax_initialization',
		    'The protocol begins with a dot on a line by itself.'));
		W0('Closed Connection to %s', $self->{'R_HOST'});
		Net::PSYC::shutdown($self);
	    }
	}
    }

    return 1;
}

# return undef if packets are incomplete
# return 0 if there maybe/are still packets in the buffer
# return the packet
sub recv () {
    my $self = shift;

    return unless ($self->{'LF'});
    return if ($self->{'I_LENGTH'} < 0 || '' eq $self->{'I_BUFFER'});

    my ($vars, $data) = parse_mmp(\$$self{'I_BUFFER'}, $self->{'LF'}, $self);

    return if (!defined($vars));

    if ($vars < 0) {
	$self->{'I_LENGTH'} = $vars;
	return;
    }

    if ($vars == 0) {
	return (-1, $data);
    }

if (!Net::PSYC::FORK) {
    if (exists $vars->{'_using_modules'}) {

	unless (ref $vars->{'_using_modules'} eq 'ARRAY') {
	    $self->negotiate($vars->{'_using_modules'})
		if $vars->{'_using_modules'};
	} else {
	    map { $_ && $self->negotiate($_) } @{$vars->{'_using_modules'}};
	}
    }
}

    return (-1, "Fatal error during receive.")
	unless ($self->trigger('receive', $vars, \$data));

    unless (exists $self->{'me'} || $self->{'L'} || !exists $vars->{'_target'}) {
	$self->{'me'} = $vars->{'_target'};
	my $r = parse_uniform($vars->{'_target'});
	if (ref $r && $r->{'host'}) {
	    dns_lookup($r->{'host'},
		       sub {
			    my $ip = shift;
			    unless ($ip) {
				W0('Could not resolve %s.', $r->{'host'});
			    }
			    W0('%s -> %s', $r->{'host'}, $ip);
			    register_host('127.0.0.1', $ip || $r->{'host'});
		       });
	} else {
	    W0('I cannot parse that target: %s. Closing connection.',
		$vars->{'_target'});
	    return -1;
	}
    }

    $vars = { %{$self->{'vars'}}, %$vars } if (each %{$self->{'vars'}});

    # TODO return -1 unless trigger().
    # TODO we have to check _context for consistency anyway! do that or someone
    # starts killing perlpsycs
    # these routing schemes are bogus. i would like to use the new ones. should
    # be easier to do nonblocking dns then. one big change
    unless (exists $vars->{'_source'}) {
	$vars->{'_source'} = $self->{'peeraddr'};
    } else {
	my $h = parse_uniform($vars->{'_context'}||$vars->{'_source'});
	unless (ref $h) {
	    W0('I cannot parse that uni: %s. Closing connection.',
		$vars->{'_context'} || $vars->{'_source'} );
	    return -1;
	}

	unless (same_host($h->{'host'}, $self->{'R_IP'})) {
	    if ($self->TRUST < 5) {
		# just dont relay
		W0('TCP: Refused packet from %s. (_source: %s)',
		    $self->{'peeraddr'}, $vars->{'_source'});
		return 0;
	    }
	} else {
	    # we will relay for you in the future
	    register_route($vars->{'_source'}, $self);
	}
    }
=onion
    if (exists $vars->{'_source_relay'} && $self->{'_options'}->{'_accept_modules'} =~ /_onion/ && $self->{'r_options'}->{'_accept_modules'} =~ /_onion/) {
	register_route($vars->{'_source_relay'}, $self);
	W("_Onion: Use $self->{'R_IP'} to route $vars->{'_source_relay'}",2);
	# remember pseudo-address to route packets back!
    }
=cut
    ####
    # FRAGMENT
    # handle fragmented data
    #if (exists $self->{'O'}->{'_understand_modules'}->{'_fragments'}
    #&& exists $vars->{'_fragment'}) {
    if (exists $vars->{'_fragment'}) {
	# {source} {logical target} {counter} [ {fragment} ]
	my $packet_id = '{'.($vars->{'_source'} || '').
			'}{'.($vars->{'_target'} || '').
			'}{'.($vars->{'_counter'} || '').'}';
	if (!exists $self->{'CACHE'}->{$packet_id}) {
	    $self->{'CACHE'}->{$packet_id} = [
		{
		    '_totalLength' => $vars->{'_totalLength'},
		    '_amount_fragments' => $vars->{'_amount_fragments'},
		    '_amount' => 0,
		},
		[]
	    ];
	}
	my $v = $self->{'CACHE'}->{$packet_id}->[0];
	my $c = $self->{'CACHE'}->{$packet_id}->[1];
	# increase the counter
	$v->{'_amount'}++ if (!$c->[$vars->{'_fragment'}]);
	#print STDERR "Fragment: $vars->{'_fragment'} (total: $vars->{'_amount_fragments'}, amount: $v->{'_amount'}, id: '$packet_id')\n";

	$c->[$vars->{'_fragment'}] = $data;
	if ($v->{'_amount'} == $v->{'_amount_fragments'}) {
	    $data = join('', @$c);
	    delete $self->{'CACHE'}->{$packet_id};
	    W1('TCP: Fragmented packet complete! length: %d', length($data));
	} else {
	    W1('TCP: Fragmented number %d', int($vars->{'_fragment'}));
	    return 0;
	}
    }
    ####
    return 0 if ($data eq '');

    W1('TCP[%s] => %s', $vars->{'_source'}, $vars->{'_target'});
    $vars->{'_INTERNAL_origin'} = $self;
    return ($vars, $data);
}

sub DESTROY {
    my $self = shift;
    $self->{'SOCKET'}->shutdown(0) if $self->{'SOCKET'};
}

# TCP listen class
package Net::PSYC::Circuit::L;

use strict;

import Net::PSYC qw(W0);

sub read () {
    my $self = shift;
    my $socket = $self->{'SOCKET'}->accept();
    my $obj = Net::PSYC::Circuit->new($socket, {
	'L' => 1,
	'R_IP' => $socket->peerhost(),
	'R_PORT' => $socket->peerport(),
	});
    return 1;
}

sub recv () { }

sub send {
    W0("\nTCP: I am listening, not sending! Dont use me that way!");
}

sub TRUST {
    W0("\nTCP: Dont TRUST() me, I'm only listening.");
}

1;