sessions/mid_event.pl

   1 #!/usr/local/bin/perl -w
   2
   3 use strict ;
   4 use lib '../lib' ;
   5
   6 BEGIN {
   7         $Stem::Vars::Env{event_loop} = 'perl' ;
   8 }
   9
  10
  11 use Stem::Event ;
  12 use Stem::Socket ;
  13 use Stem::AsyncIO ;
  14
  15 use Data::Dumper ;
  16 use Getopt::Long ;
  17
  18 my $opts_ok = GetOptions(
  19         \my %opts,
  20         'server_port=s',
  21         'upper_port=s',
  22         'reverse_port=s',
  23         'verbose|v',
  24         'help|h',
  25 ) ;
  26
  27 usage() unless $opts_ok ;
  28 usage() if $opts{help} ;
  29
  30 my %backend_ports = (
  31
  32         'reverse'       => $opts{reverse_port} || 8888,
  33         'upper'         => $opts{upper_port} || 8889,
  34 ) ;
  35
  36 # this controls the order of requests to the backends.
  37
  38 my @backend_ids = sort keys %backend_ports ;
  39
  40 my $listen = init_server( $opts{server_port} || 8887 ) ;
  41
  42 Stem::Event::start_loop() ;
  43
  44 exit ;
  45
  46 # create the listen socket for the server side of the middle layer.
  47
  48 sub init_server {
  49
  50         my( $port ) = @_ ;
  51
  52 # create the middle layer listen socket
  53
  54         my $listen = Stem::Socket->new(
  55                 object  => bless( {
  56                 }, __PACKAGE__),
  57                 method  => 'client_connected',
  58                 port    => $port,
  59                 server  => 1,
  60         ) ;
  61
  62         die "can't listen on $port: $listen" unless ref $listen ;
  63
  64         return $listen ;
  65 }
  66
  67 # this is called when the server has accepted a socket connection
  68
  69 sub client_connected {
  70
  71         my( $obj, $socket ) = @_ ;
  72
  73 # create the session object
  74
  75         my $self = bless {}, __PACKAGE__ ;
  76
  77 # create and save the async io object for the client
  78
  79         my $async = Stem::AsyncIO->new(
  80                 object  => $self,
  81                 fh      => $socket,
  82                 read_method     => 'client_read_data',
  83                 send_data_on_close => 1,
  84         ) ;
  85         ref $async or die "can't create Async: $async" ;
  86         $self->{client_async} = $async ;
  87
  88 # store a copy of the backend as we shift them out
  89
  90         $self->{backend_ids} = [ @backend_ids ] ;
  91
  92 }
  93
  94 # this is called when all the data from client has been read.
  95
  96 sub client_read_data {
  97
  98         my( $self, $data ) = @_ ;
  99
 100         print "Client read [${$data}]\n"  if $opts{verbose} ;
 101
 102 # store the client data (a ref is passed in)
 103
 104         $self->{'client_data'} = ${$data} ;
 105
 106 # connect to the first backend server
 107
 108         my $backend_id = shift( @{$self->{backend_ids}} ) ;
 109
 110         $self->connect_to_backend( $backend_id ) ;
 111 }
 112
 113 # this connects the session to one of the backends
 114
 115 sub connect_to_backend {
 116
 117         my( $self, $id ) = @_ ;
 118
 119 # connect to the backend with this id and its port and save the
 120 # connect object
 121
 122         my $connect = Stem::Socket->new(
 123                 object  => $self,
 124                 id      => $id,
 125                 port    => $backend_ports{ $id },
 126                 method  => 'backend_connected',
 127         ) ;
 128
 129         ref $connect or die "can't create Socket: $connect" ;
 130         $self->{connect}{$id} = $connect ;
 131 }
 132
 133 # this is called when a backend end connection succeeds
 134
 135 sub backend_connected {
 136
 137         my( $self, $socket, $id ) = @_ ;
 138
 139 # delete and shutdown the connect object as we no longer need it
 140
 141         my $connect = delete $self->{connect}{$id} ;
 142         $connect->shut_down() ;
 143
 144 # create and save an async i/o object for this backend
 145
 146         my $async = Stem::AsyncIO->new(
 147                 object  => $self,
 148                 id      => $id,
 149                 fh      => $socket,
 150                 read_method     => 'backend_read_data',
 151                 send_data_on_close => 1,
 152         ) ;
 153         ref $async or die "can't create Async: $async" ;
 154         $self->{async}{$id} = $async ;
 155
 156 # write the client data to the back end. no more data will follow.
 157
 158         $async->final_write( \$self->{client_data} ) ;
 159 }
 160
 161 # this is called when we have read all the data from the backend
 162
 163 sub backend_read_data {
 164
 165         my( $self, $data, $id ) = @_ ;
 166
 167         print "Backend $id READ [${$data}]\n" if $opts{verbose} ;
 168
 169 # save the backend data (we are passed a ref)
 170
 171         $self->{backend_data}{$id} = ${$data} ;
 172
 173 # delete and shutdown the async i/o for the backend since we don't
 174 # need it anymore
 175
 176         my $async = delete $self->{async}{$id} ;
 177         $async->shut_down() ;
 178
 179 # do the next backend in the list.  this is a simple way to handle
 180 # sequential backends we use the backend_ids array in the session
 181 # object to track which backends we have not used yet.
 182
 183         if ( my $backend_id = shift( @{$self->{backend_ids}} ) ) {
 184
 185 # connect to the next backend server.
 186
 187                 $self->connect_to_backend( $backend_id ) ;
 188                 return ;
 189         }
 190
 191 # no more backends so we return the joined backend data to the client.
 192
 193 # delete the async so we don't keep a ref to it around. this will
 194 # allow for self cleanup when it is done with the final write to the
 195 # client.
 196
 197         $async = delete $self->{client_async} ;
 198         $async->final_write(
 199                 join( '',  @{$self->{backend_data}}{ @backend_ids } )
 200         ) ;
 201 }
 202
 203 sub usage {
 204
 205         my ( $error ) = @_ ;
 206
 207         $error ||= '' ;
 208         die <<DIE ;
 209 $error
 210 usage: $0 [--help|h] [--upper_port <port>] [--reverse_port <port>]
 211         [--server_port <port>] [--v|--verbose]
 212
 213         upper_port <port>       Set the port for the middleware server
 214                                 (default is 8888)
 215         upper_port <port>       Set the port for the upper case server
 216                                 (default is 8888)
 217         reverse_port <port>     Set the port for the string reverse server
 218                                 (default is 8889)
 219         verbose                 Set verbose mode
 220         help | h                Print this help text
 221 DIE
 222
 223 }
 224
 225 # this destroy can be uncommented to see the actual destruction of the
 226 # various obects in this script.
 227
 228 # DESTROY {
 229 #       my( $self ) = @_ ;
 230 #       print "DEST [$self]\n" ;
 231 # }