perl/lib/Thrasher/Socket.pm

   1 package Thrasher::Socket;
   2 use strict;
   3 use warnings;
   4
   5 =pod
   6
   7 =head1 NAME
   8
   9 Thrasher::Socket - manage the socket connection to the XMPP server.
  10
  11 =head1 DESCRIPTION
  12
  13 Thrasher::Socket manages the connection to the XMPP server, and
  14 once the connection is successfully established, feeds the socket
  15 into the Component.
  16
  17 =cut
  18
  19 use POSIX;
  20 use Socket;
  21 use IO::Socket;
  22 use Thrasher::Log qw(log debug);
  23
  24 sub new {
  25     my $class = shift;
  26     my $self = {};
  27
  28     $self->{host} = shift;
  29     $self->{port} = shift;
  30     $self->{event_loop} = shift;
  31
  32     bless $self, $class;
  33     return $self;
  34 }
  35
  36 # Returns the function to feed to the component to write to
  37 # this socket.
  38 sub write_function {
  39     my $self = shift;
  40     my $writer = sub {
  41         debug("OUT: " . join('', @_));
  42
  43         $self->write(@_);
  44     };
  45     return $writer;
  46 }
  47
  48 sub write {
  49     my $self = shift;
  50
  51     my $socket = $self->{socket};
  52     for my $data (@_) {
  53         print $socket $data;
  54     }
  55     flush $socket;
  56 }
  57
  58 # dies on failure
  59 sub connect {
  60     my $self = shift;
  61     if ($self->{connected}) {
  62         return;
  63     }
  64
  65     my $proto = getprotobyname('tcp');
  66     my $iaddr = inet_aton($self->{host});
  67     my $paddr = sockaddr_in($self->{port}, $iaddr);
  68
  69     my $socket;
  70
  71     log("Beginning socket connection to $self->{host}:$self->{port}...");
  72     socket($socket, PF_INET, SOCK_STREAM, $proto)
  73         or die "SOCK_CREATE: socket failed to create";
  74     connect($socket, $paddr)
  75         or die "SOCK_CONNECT: socket failed to connect";
  76     log("connected.");
  77
  78     binmode($socket, ':encoding(utf8)');
  79
  80     # No blocking here
  81     fcntl($socket, F_SETFL, O_NONBLOCK);
  82
  83     $self->{socket}    = $socket;
  84     $self->{connected} = 1;
  85 }
  86
  87 sub establish_fd_watch {
  88     my $self = shift;
  89
  90     $self->{watch_handle} = $self->{event_loop}->add_fd_watch
  91         ($self->fileno, $Thrasher::EventLoop::IN,
  92          $self->{read_closure});
  93 }
  94
  95 sub terminate_fd_watch {
  96     my $self = shift;
  97     if (!exists($self->{watch_handle})) {
  98         return;
  99     }
 100
 101     $self->{event_loop}->remove_fd_watch($self->{watch_handle});
 102     delete($self->{watch_handle});
 103 }
 104
 105 sub get_socket {
 106     my $self = shift;
 107     if (!$self->{connected}) {
 108         die "No socket to get.";
 109     }
 110
 111     return $self->{socket};
 112 }
 113
 114 sub read {
 115     my $self = shift;
 116
 117     if (!$self->{connected}) {
 118         die('Trying to read from an unconnected socket.');
 119     }
 120
 121     my $value;
 122     my $return = sysread($self->{socket}, $value, 65535);
 123     if ($return) {
 124         return $value;
 125     }
 126     # For now restart on all read errors--dunno what state changes we
 127     # missed reading. Might eventually be able to reconnect only the
 128     # component w/o reconnecting everyone to the public IM network.
 129     elsif (! defined($return)) {
 130         if ($!{'EAGAIN'}) {
 131             # Not ready yet. That's OK--return undef and try again later.
 132             return;
 133         }
 134         else {
 135             die("Error reading from component socket: $!");
 136         }
 137     }
 138     else {
 139         die('EOF when reading from component socket!');
 140     }
 141 }
 142
 143 sub close {
 144     my $self = shift;
 145     $self->{socket}->close;
 146     $self->terminate_fd_watch;
 147     delete $self->{connected};
 148     delete $self->{socket};
 149 }
 150
 151 sub fileno {
 152     my $self = shift;
 153     return $self->{socket}->fileno;
 154 }
 155
 156 sub eof {
 157     my $self = shift;
 158     return $self->{socket}->eof;
 159 }
 160
 161 1;