[urisagit/Stem.git] / lib / Stem / SockMsg.pm

#  File: Stem/SockMsg.pm

#  This file is part of Stem.
#  Copyright (C) 1999, 2000, 2001 Stem Systems, Inc.

#  Stem 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.

#  Stem 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 Stem; if not, write to the Free Software
#  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

#  For a license to use the Stem under conditions other than those
#  described here, to purchase support for this software, or to purchase a
#  commercial warranty contract, please contact Stem Systems at:

#       Stem Systems, Inc.		781-643-7504
#  	79 Everett St.			info@stemsystems.com
#  	Arlington, MA 02474
#  	USA

package Stem::SockMsg ;

use strict ;

use Data::Dumper ;

use Stem::Socket ;
use Stem::Trace 'log' => 'stem_status', 'sub' => 'TraceStatus' ;
use Stem::Trace 'log' => 'stem_error' , 'sub' => 'TraceError' ;
use Stem::Route qw( :cell ) ;
use base 'Stem::Cell' ;

use Stem::Debug qw( dump_data dump_socket ) ;


my $attr_spec = [

	{
		'name'		=> 'reg_name',
		'help'		=> <<HELP,
The registration name for this Cell
HELP
	},

	{
		'name'		=> 'host',
		'env'		=> 'host',
		'help'		=> <<HELP,
Host address to listen on or connect to
HELP
	},

	{
		'name'		=> 'port',
		'env'		=> 'port',
		'required'	=> 1,
		'help'		=> <<HELP,
Port address to listen on or connect to
HELP
	},

	{
		'name'		=> 'server',
		'type'		=> 'boolean',
		'help'		=> <<HELP,
Mark this Cell as a server (listens for connections)
HELP
	},

	{
		'name'		=> 'connect_now',
		'type'		=> 'boolean',
		'help'		=> <<HELP,
Connect upon Cell creation
HELP
	},

	{
		'name'		=> 'status_addr',
		'type'		=> 'address',
		'help'		=> <<HELP,
Send status (connect/disconnect) messages to this address.
HELP
	},

	{
		'name'		=> 'sync',
		'type'		=> 'boolean',
		'default'	=> 0,
		'help'		=> <<HELP,
Mark this as a synchronously connecting socket. Default is asyncronous
connections. In both cases the same method callbacks are used.
HELP
	},

	{
		'name'		=> 'log_name',
		'help'		=> <<HELP,
Log to send connection status to
HELP
	},

	{
		'name'		=> 'cell_attr',
		'class'		=> 'Stem::Cell',
		'help'		=> <<HELP,
Argument list passed to Stem::Cell for this Cell
HELP
	},

] ;

#my $listener ;


sub new {

	my( $class ) = shift ;

	my $self = Stem::Class::parse_args( $attr_spec, @_ ) ;
	return $self unless ref $self ;

	if ( $self->{'server'} ) {
		my $listen_obj = Stem::Socket->new( 
				'object'	=> $self,
				'host'		=> $self->{'host'},
				'port'		=> $self->{'port'},
				'server'	=> 1,
		) ;

		return $listen_obj unless ref $listen_obj ;

		my $host_text = $self->{'host'} ;

		$host_text = 'localhost' unless defined $host_text ;

		my $info = <<INFO ;
SockMsg
Type:	server
Local:	$host_text:$self->{'port'}
INFO

		$self->cell_info( $info ) ;

		$self->{'listen_obj'} = $listen_obj ;

#print "LISTEN $listen_obj\n" ;
#$listener = $listen_obj ;

		$self->cell_activate() ;
	}
	elsif ( $self->{'connect_now'} ) {

#print "NOW\n" ;

		$self->connect() ;
	}

	$self->cell_set_args(
			'host'		=> $self->{'host'},
			'port'		=> $self->{'port'},
			'server'	=> $self->{'server'},
	) ;

#print  "Sock\n", Dumper( $self ) ;

	return( $self ) ;
}

sub connect {

	my( $self ) = @_ ;

#print "MODE [$self->{connecting}]\n" ;

#	return if $self->{connecting}++ ;

	my $host = $self->cell_get_args( 'host' ) || $self->{'host'} ;
	my $port = $self->cell_get_args( 'port' ) || $self->{'port'} ;
	my $sync = $self->cell_get_args( 'sync' ) || $self->{'sync'} ;

########################
########################
## handle connect timeouts
########################
########################

#TraceStatus "Connecting to $host:$port" ;

	my $sock_obj = Stem::Socket->new( 
			'object'	=> $self,
			'host'		=> $host,
			'port'		=> $port,
			'sync'		=> $sync,
	) ;

	return $sock_obj unless ref $sock_obj ;

	$self->{'sock_obj'} = $sock_obj ;

	return ;
}

sub connected {

	my( $self, $connected_sock ) = @_ ;

#print "CONNECTED\n" ;

	$self->{connected} = 1 ;

	$self->send_status_msg( 'connected' ) ;

	my $type = $self->{'sock_obj'} ?
			$self->{'sock_obj'}->type() :
			'sync connected' ;

	my $info = sprintf( <<INFO,
SockMsg connected
Type:	$type
Local:	%s:%d
Remote:	%s:%d
INFO
				$connected_sock->sockhost(),
				$connected_sock->sockport(),
				$connected_sock->peerhost(),
				$connected_sock->peerport(),
	) ;

	TraceStatus "\n$info" ;

#print "\n$info" ;

	if ( my $log_name = $self->{ 'log_name' } ) {

#print "MSG LOG\n" ;

		Stem::Log::Entry->new(
				'logs'	=> $log_name,
				'text'	=> "Connected\n$info",
		) ;
	}

	$self->cell_set_args(
			'fh'		=> $connected_sock,
			'aio_args'	=>
				[ 'fh'	=> $connected_sock ],
			'info'		=> $info,
	) ;

	my $err = $self->cell_trigger() ;
#	print "TRIGGER ERR [$err]\n" unless ref $err ;
}

# this method is called after the cell is triggered. this cell can be
# the original cell or a cloned one.

sub triggered_cell {

	my( $self ) = @_ ;

#print "SockMsg triggered\n" ;
	return if $self->{'connected'} ;
	return if $self->{'server'} ;

#	return "SockMsg: can't connect a server socket" if $self->{'server'} ;

	return $self->connect() ;
}

# we handle the socket close method directly here so we can reconnect
# if needed. the other async method callbacks are in Cell.pm

sub async_closed {

	my( $self ) = @_ ;

# reconnect stuff. should be in Socket.pm

#	my $sock = $self->cell_get_args( 'fh' ) ;
#	$sock->close() ;
#print "Sock MSG: closed name $self->{'reg_name'}\n" ;
#	$self->{'sock_obj'}->connect_to() ;

	$self->send_status_msg( 'disconnected' ) ;

	if ( my $log_name = $self->{ 'log_name' } ) {

		Stem::Log::Entry->new(
				'logs'	=> $log_name,
				'text'	=> "Closed\n$self->{'info'}",
		)
	}

#	TraceStatus "Disconnected" ;

	$self->cell_set_args( 'info' => 'SockMsg disconnected' ) ;

######################
######################
# add support for reconnect.
# it has a flag, delay, retry count.
######################
######################

	$self->shut_down() ;
}

sub shut_down {

	my( $self ) = @_ ;

#print "SOCKMSG SHUT $self\n", caller(), "\n", dump_data $self ;

	$self->cell_shut_down() ;

	unless ( $self->{'connected'} ) {

use Carp 'cluck' ;
#cluck "SOCKMSG SHUT SERVER $self\n" ;

		my $sock_obj = $self->{'sock_obj'} ;

		$sock_obj->shut_down() ;
	}
}

sub send_status_msg {

	my( $self, $status ) = @_ ;

	my $status_addr = $self->{status_addr} or return ;

	my $status_msg = Stem::Msg->new(
		to	=> $status_addr,
		from	=> $self->cell_from_addr(),
		type	=> 'status',
		data	=> {
			status	=> $status,
		},
	) ;

	$status_msg->dispatch() ;
}


sub DESTROY {
 	my ( $self ) = @_ ;

# print "SOCKMSG DESTROY", caller(), "\n" ;

#print $self->_dump( "DESTROY") ;
}


# sub IO::Socket::INET::DESTROY {
#  	my ( $self ) = @_ ;

# #	print "IO::DESTROY\n", dump_socket( $self ) ;

# #warn "L $listener - S $self\n" if $listener == $self ;

# # print "SOCKMSG DESTROY", caller(), "\n" ;
# #cluck "IO::DESTROY $self\n" ;
# }

1 ;
Commit	Line	Data
4536f655	1	# File: Stem/SockMsg.pm
	2
	3	# This file is part of Stem.
	4	# Copyright (C) 1999, 2000, 2001 Stem Systems, Inc.
	5
	6	# Stem is free software; you can redistribute it and/or modify
	7	# it under the terms of the GNU General Public License as published by
	8	# the Free Software Foundation; either version 2 of the License, or
	9	# (at your option) any later version.
	10
	11	# Stem is distributed in the hope that it will be useful,
	12	# but WITHOUT ANY WARRANTY; without even the implied warranty of
	13	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	14	# GNU General Public License for more details.
	15
	16	# You should have received a copy of the GNU General Public License
	17	# along with Stem; if not, write to the Free Software
	18	# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
	19
	20	# For a license to use the Stem under conditions other than those
	21	# described here, to purchase support for this software, or to purchase a
	22	# commercial warranty contract, please contact Stem Systems at:
	23
	24	# Stem Systems, Inc. 781-643-7504
	25	# 79 Everett St. info@stemsystems.com
	26	# Arlington, MA 02474
	27	# USA
	28
	29	package Stem::SockMsg ;
	30
	31	use strict ;
	32
	33	use Data::Dumper ;
	34
	35	use Stem::Socket ;
	36	use Stem::Trace 'log' => 'stem_status', 'sub' => 'TraceStatus' ;
	37	use Stem::Trace 'log' => 'stem_error' , 'sub' => 'TraceError' ;
	38	use Stem::Route qw( :cell ) ;
	39	use base 'Stem::Cell' ;
	40
	41	use Stem::Debug qw( dump_data dump_socket ) ;
	42
	43
	44	my $attr_spec = [
	45
	46	{
	47	'name' => 'reg_name',
	48	'help' => <<HELP,
	49	The registration name for this Cell
	50	HELP
	51	},
	52
	53	{
	54	'name' => 'host',
	55	'env' => 'host',
	56	'help' => <<HELP,
	57	Host address to listen on or connect to
	58	HELP
	59	},
	60
	61	{
	62	'name' => 'port',
	63	'env' => 'port',
	64	'required' => 1,
65	'help' => <<HELP,
66	Port address to listen on or connect to
67	HELP
68	},
69
70	{
71	'name' => 'server',
72	'type' => 'boolean',
73	'help' => <<HELP,
74	Mark this Cell as a server (listens for connections)
75	HELP
76	},
77
78	{
79	'name' => 'connect_now',
80	'type' => 'boolean',
81	'help' => <<HELP,
82	Connect upon Cell creation
83	HELP
84	},
85
86	{
87	'name' => 'status_addr',
88	'type' => 'address',
89	'help' => <<HELP,
90	Send status (connect/disconnect) messages to this address.
91	HELP
92	},
93
94	{
95	'name' => 'sync',
96	'type' => 'boolean',
97	'default' => 0,
98	'help' => <<HELP,
99	Mark this as a synchronously connecting socket. Default is asyncronous
100	connections. In both cases the same method callbacks are used.
101	HELP
102	},
103
104	{
105	'name' => 'log_name',
106	'help' => <<HELP,
107	Log to send connection status to
108	HELP
109	},
110
111	{
112	'name' => 'cell_attr',
113	'class' => 'Stem::Cell',
114	'help' => <<HELP,
115	Argument list passed to Stem::Cell for this Cell
116	HELP
117	},
118
119	] ;
120
121	#my $listener ;
122
123
124	sub new {
125
126	my( $class ) = shift ;
127
128	my $self = Stem::Class::parse_args( $attr_spec, @_ ) ;
129	return $self unless ref $self ;
130
131	if ( $self->{'server'} ) {
132	my $listen_obj = Stem::Socket->new(
133	'object' => $self,
134	'host' => $self->{'host'},
135	'port' => $self->{'port'},
136	'server' => 1,
137	) ;
138
139	return $listen_obj unless ref $listen_obj ;
140
141	my $host_text = $self->{'host'} ;
142
143	$host_text = 'localhost' unless defined $host_text ;
144
145	my $info = <<INFO ;
146	SockMsg
147	Type: server
148	Local: $host_text:$self->{'port'}
149	INFO
150
151	$self->cell_info( $info ) ;
152
153	$self->{'listen_obj'} = $listen_obj ;
154
155	#print "LISTEN $listen_obj\n" ;
156	#$listener = $listen_obj ;
157
158	$self->cell_activate() ;
159	}
160	elsif ( $self->{'connect_now'} ) {
161
bd113ec4	162	#print "NOW\n" ;
bd113ec4	163
4536f655	164	$self->connect() ;
	165	}
	166
	167	$self->cell_set_args(
	168	'host' => $self->{'host'},
	169	'port' => $self->{'port'},
	170	'server' => $self->{'server'},
	171	) ;
	172
	173	#print "Sock\n", Dumper( $self ) ;
	174
	175	return( $self ) ;
	176	}
	177
	178	sub connect {
	179
	180	my( $self ) = @_ ;
	181
	182	#print "MODE [$self->{connecting}]\n" ;
	183
	184	# return if $self->{connecting}++ ;
	185
	186	my $host = $self->cell_get_args( 'host' ) \|\| $self->{'host'} ;
	187	my $port = $self->cell_get_args( 'port' ) \|\| $self->{'port'} ;
	188	my $sync = $self->cell_get_args( 'sync' ) \|\| $self->{'sync'} ;
	189
	190	########################
	191	########################
	192	## handle connect timeouts
	193	########################
	194	########################
	195
	196	#TraceStatus "Connecting to $host:$port" ;
	197
	198	my $sock_obj = Stem::Socket->new(
	199	'object' => $self,
	200	'host' => $host,
	201	'port' => $port,
	202	'sync' => $sync,
	203	) ;
	204
	205	return $sock_obj unless ref $sock_obj ;
	206
	207	$self->{'sock_obj'} = $sock_obj ;
	208
	209	return ;
	210	}
	211
	212	sub connected {
	213
	214	my( $self, $connected_sock ) = @_ ;
	215
	216	#print "CONNECTED\n" ;
	217
	218	$self->{connected} = 1 ;
	219
	220	$self->send_status_msg( 'connected' ) ;
	221
	222	my $type = $self->{'sock_obj'} ?
	223	$self->{'sock_obj'}->type() :
	224	'sync connected' ;
	225
	226	my $info = sprintf( <<INFO,
	227	SockMsg connected
228	Type: $type
229	Local: %s:%d
230	Remote: %s:%d
231	INFO
232	$connected_sock->sockhost(),
233	$connected_sock->sockport(),
234	$connected_sock->peerhost(),
235	$connected_sock->peerport(),
236	) ;
237
238	TraceStatus "\n$info" ;
239
bd113ec4	240	#print "\n$info" ;
bd113ec4	241
4536f655	242	if ( my $log_name = $self->{ 'log_name' } ) {
	243
	244	#print "MSG LOG\n" ;
	245
	246	Stem::Log::Entry->new(
	247	'logs' => $log_name,
	248	'text' => "Connected\n$info",
	249	) ;
	250	}
	251
	252	$self->cell_set_args(
	253	'fh' => $connected_sock,
	254	'aio_args' =>
	255	[ 'fh' => $connected_sock ],
	256	'info' => $info,
	257	) ;
	258
	259	my $err = $self->cell_trigger() ;
	260	# print "TRIGGER ERR [$err]\n" unless ref $err ;
	261	}
	262
	263	# this method is called after the cell is triggered. this cell can be
	264	# the original cell or a cloned one.
	265
	266	sub triggered_cell {
	267
	268	my( $self ) = @_ ;
	269
	270	#print "SockMsg triggered\n" ;
bd113ec4	271	return if $self->{'connected'} ;
4536f655	272	return if $self->{'server'} ;
	273
	274	# return "SockMsg: can't connect a server socket" if $self->{'server'} ;
	275
	276	return $self->connect() ;
	277	}
	278
	279	# we handle the socket close method directly here so we can reconnect
	280	# if needed. the other async method callbacks are in Cell.pm
	281
	282	sub async_closed {
	283
	284	my( $self ) = @_ ;
	285
	286	# reconnect stuff. should be in Socket.pm
	287
	288	# my $sock = $self->cell_get_args( 'fh' ) ;
	289	# $sock->close() ;
	290	#print "Sock MSG: closed name $self->{'reg_name'}\n" ;
	291	# $self->{'sock_obj'}->connect_to() ;
	292
	293	$self->send_status_msg( 'disconnected' ) ;
	294
	295	if ( my $log_name = $self->{ 'log_name' } ) {
	296
	297	Stem::Log::Entry->new(
	298	'logs' => $log_name,
	299	'text' => "Closed\n$self->{'info'}",
	300	)
	301	}
	302
	303	# TraceStatus "Disconnected" ;
	304
	305	$self->cell_set_args( 'info' => 'SockMsg disconnected' ) ;
	306
	307	######################
	308	######################
	309	# add support for reconnect.
	310	# it has a flag, delay, retry count.
	311	######################
	312	######################
	313
	314	$self->shut_down() ;
	315	}
	316
	317	sub shut_down {
	318
	319	my( $self ) = @_ ;
	320
	321	#print "SOCKMSG SHUT $self\n", caller(), "\n", dump_data $self ;
	322
	323	$self->cell_shut_down() ;
	324
	325	unless ( $self->{'connected'} ) {
	326
	327	use Carp 'cluck' ;
	328	#cluck "SOCKMSG SHUT SERVER $self\n" ;
	329
	330	my $sock_obj = $self->{'sock_obj'} ;
	331
	332	$sock_obj->shut_down() ;
	333	}
	334	}
	335
336	sub send_status_msg {
337
338	my( $self, $status ) = @_ ;
339
340	my $status_addr = $self->{status_addr} or return ;
341
342	my $status_msg = Stem::Msg->new(
343	to => $status_addr,
344	from => $self->cell_from_addr(),
345	type => 'status',
346	data => {
347	status => $status,
348	},
349	) ;
350
351	$status_msg->dispatch() ;
352	}
353
354
355
356	sub DESTROY {
357	my ( $self ) = @_ ;
358
359	# print "SOCKMSG DESTROY", caller(), "\n" ;
360
361	#print $self->_dump( "DESTROY") ;
362	}
363
364
365	# sub IO::Socket::INET::DESTROY {
366	# my ( $self ) = @_ ;
367
368	# # print "IO::DESTROY\n", dump_socket( $self ) ;
369
370	# #warn "L $listener - S $self\n" if $listener == $self ;
371
372	# # print "SOCKMSG DESTROY", caller(), "\n" ;
373	# #cluck "IO::DESTROY $self\n" ;
374	# }
375
376	1 ;