src/perlmods/OpenSRF/Transport/SlimJabber/XMPPReader.pm

   1 package OpenSRF::Transport::SlimJabber::XMPPReader;
   2 use strict; use warnings;
   3 use XML::Parser;
   4 use Fcntl qw(F_GETFL F_SETFL O_NONBLOCK);
   5 use Time::HiRes qw/time/;
   6 use OpenSRF::Transport::SlimJabber::XMPPMessage;
   7 use OpenSRF::Utils::Logger qw/$logger/;
   8
   9 # -----------------------------------------------------------
  10 # Connect, disconnect, and authentication messsage templates
  11 # -----------------------------------------------------------
  12 use constant JABBER_CONNECT =>
  13     "<stream:stream to='%s' xmlns='jabber:client' xmlns:stream='http://etherx.jabber.org/streams'>";
  14
  15 use constant JABBER_BASIC_AUTH =>
  16     "<iq id='123' type='set'><query xmlns='jabber:iq:auth'>" .
  17     "<username>%s</username><password>%s</password><resource>%s</resource></query></iq>";
  18
  19 use constant JABBER_DISCONNECT => "</stream:stream>";
  20
  21
  22 # -----------------------------------------------------------
  23 # XMPP Stream states
  24 # -----------------------------------------------------------
  25 use constant DISCONNECTED   => 1;
  26 use constant CONNECT_RECV   => 2;
  27 use constant CONNECTED      => 3;
  28
  29
  30 # -----------------------------------------------------------
  31 # XMPP Message states
  32 # -----------------------------------------------------------
  33 use constant IN_NOTHING => 1;
  34 use constant IN_BODY    => 2;
  35 use constant IN_THREAD  => 3;
  36 use constant IN_STATUS  => 4;
  37
  38
  39 # -----------------------------------------------------------
  40 # Constructor, getter/setters
  41 # -----------------------------------------------------------
  42 sub new {
  43     my $class = shift;
  44     my $socket = shift;
  45
  46     my $self = bless({}, $class);
  47
  48     $self->{queue} = [];
  49     $self->{stream_state} = DISCONNECTED;
  50     $self->{xml_state} = IN_NOTHING;
  51     $self->socket($socket);
  52
  53     my $p = new XML::Parser(Handlers => {
  54         Start => \&start_element,
  55         End   => \&end_element,
  56         Char  => \&characters,
  57     });
  58
  59     $self->parser($p->parse_start); # create a push parser
  60     $self->parser->{_parent_} = $self;
  61     $self->{message} = OpenSRF::Transport::SlimJabber::XMPPMessage->new;
  62     return $self;
  63 }
  64
  65 sub push_msg {
  66     my($self, $msg) = @_;
  67     push(@{$self->{queue}}, $msg) if $msg;
  68 }
  69
  70 sub next_msg {
  71     my $self = shift;
  72     return shift @{$self->{queue}};
  73 }
  74
  75 sub peek_msg {
  76     my $self = shift;
  77     return (@{$self->{queue}} > 0);
  78 }
  79
  80 sub parser {
  81     my($self, $parser) = @_;
  82     $self->{parser} = $parser if $parser;
  83     return $self->{parser};
  84 }
  85
  86 sub socket {
  87     my($self, $socket) = @_;
  88     $self->{socket} = $socket if $socket;
  89     return $self->{socket};
  90 }
  91
  92 sub stream_state {
  93     my($self, $stream_state) = @_;
  94     $self->{stream_state} = $stream_state if $stream_state;
  95     return $self->{stream_state};
  96 }
  97
  98 sub xml_state {
  99     my($self, $xml_state) = @_;
 100     $self->{xml_state} = $xml_state if $xml_state;
 101     return $self->{xml_state};
 102 }
 103
 104 sub message {
 105     my($self, $message) = @_;
 106     $self->{message} = $message if $message;
 107     return $self->{message};
 108 }
 109
 110
 111 # -----------------------------------------------------------
 112 # Stream and connection handling methods
 113 # -----------------------------------------------------------
 114
 115 sub connect {
 116     my($self, $domain, $username, $password, $resource) = @_;
 117
 118     $self->send(sprintf(JABBER_CONNECT, $domain));
 119     $self->wait(10);
 120
 121     unless($self->{stream_state} == CONNECT_RECV) {
 122         $logger->error("No initial XMPP response from server");
 123         return 0;
 124     }
 125
 126     $self->send(sprintf(JABBER_BASIC_AUTH, $username, $password, $resource));
 127     $self->wait(10);
 128
 129     unless($self->connected) {
 130         $logger->error('XMPP connect failed');
 131         return 0;
 132     }
 133
 134     return 1;
 135 }
 136
 137 sub disconnect {
 138     my $self = shift;
 139     $self->send(JABBER_DISCONNECT);
 140     shutdown($self->socket, 2);
 141     close($self->socket);
 142 }
 143
 144 # -----------------------------------------------------------
 145 # returns true if this stream is connected to the server
 146 # -----------------------------------------------------------
 147 sub connected {
 148     my $self = shift;
 149     return ($self->tcp_connected and $self->{stream_state} == CONNECTED);
 150 }
 151
 152 # -----------------------------------------------------------
 153 # returns true if the socket is connected
 154 # -----------------------------------------------------------
 155 sub tcp_connected {
 156     my $self = shift;
 157     return ($self->socket and $self->socket->connected);
 158 }
 159
 160 # -----------------------------------------------------------
 161 # sends pre-formated XML
 162 # -----------------------------------------------------------
 163 sub send {
 164     my($self, $xml) = @_;
 165     $self->{socket}->print($xml);
 166 }
 167
 168 # -----------------------------------------------------------
 169 # Puts a file handle into blocking mode
 170 # -----------------------------------------------------------
 171 sub set_block {
 172     my $fh = shift;
 173     my  $flags = fcntl($fh, F_GETFL, 0);
 174     $flags &= ~O_NONBLOCK;
 175     fcntl($fh, F_SETFL, $flags);
 176 }
 177
 178
 179 # -----------------------------------------------------------
 180 # Puts a file handle into non-blocking mode
 181 # -----------------------------------------------------------
 182 sub set_nonblock {
 183     my $fh = shift;
 184     my  $flags = fcntl($fh, F_GETFL, 0);
 185     fcntl($fh, F_SETFL, $flags | O_NONBLOCK);
 186 }
 187
 188
 189 sub wait {
 190     my($self, $timeout) = @_;
 191
 192     return $self->next_msg if $self->peek_msg;
 193
 194     $timeout ||= 0;
 195     $timeout = undef if $timeout < 0;
 196     my $socket = $self->{socket};
 197
 198     set_block($socket);
 199
 200     # build the select readset
 201     my $infile = '';
 202     vec($infile, $socket->fileno, 1) = 1;
 203     return undef unless select($infile, undef, undef, $timeout);
 204
 205     # now slurp the data off the socket
 206     my $buf;
 207     my $read_size = 1024;
 208     while(my $n = sysread($socket, $buf, $read_size)) {
 209         $self->{parser}->parse_more($buf) if $buf;
 210         if($n < $read_size or $self->peek_msg) {
 211             set_block($socket);
 212             last;
 213         }
 214         set_nonblock($socket);
 215     }
 216
 217     return $self->next_msg;
 218 }
 219
 220 # -----------------------------------------------------------
 221 # Waits up to timeout seconds for a fully-formed XMPP
 222 # message to arrive.  If timeout is < 0, waits indefinitely
 223 # -----------------------------------------------------------
 224 sub wait_msg {
 225     my($self, $timeout) = @_;
 226     my $xml;
 227
 228     $timeout = 0 unless defined $timeout;
 229
 230     if($timeout < 0) {
 231         while(1) {
 232             return $xml if $xml = $self->wait($timeout);
 233         }
 234
 235     } else {
 236         while($timeout >= 0) {
 237             my $start = time;
 238             return $xml if $xml = $self->wait($timeout);
 239             $timeout -= time - $start;
 240         }
 241     }
 242
 243     return undef;
 244 }
 245
 246
 247 # -----------------------------------------------------------
 248 # SAX Handlers
 249 # -----------------------------------------------------------
 250
 251
 252 sub start_element {
 253     my($parser, $name, %attrs) = @_;
 254     my $self = $parser->{_parent_};
 255
 256     if($name eq 'message') {
 257
 258         my $msg = $self->{message};
 259         $msg->{to} = $attrs{'to'};
 260         $msg->{from} = $attrs{router_from} if $attrs{router_from};
 261         $msg->{from} = $attrs{from} unless $msg->{from};
 262         $msg->{osrf_xid} = $attrs{'osrf_xid'};
 263         $msg->{type} = $attrs{type};
 264
 265     } elsif($name eq 'body') {
 266         $self->{xml_state} = IN_BODY;
 267
 268     } elsif($name eq 'thread') {
 269         $self->{xml_state} = IN_THREAD;
 270
 271     } elsif($name eq 'stream:stream') {
 272         $self->{stream_state} = CONNECT_RECV;
 273
 274     } elsif($name eq 'iq') {
 275         if($attrs{type} and $attrs{type} eq 'result') {
 276             $self->{stream_state} = CONNECTED;
 277         }
 278
 279     } elsif($name eq 'status') {
 280         $self->{xml_state } = IN_STATUS;
 281
 282     } elsif($name eq 'stream:error') {
 283         $self->{stream_state} = DISCONNECTED;
 284
 285     } elsif($name eq 'error') {
 286         $self->{message}->{err_type} = $attrs{'type'};
 287         $self->{message}->{err_code} = $attrs{'code'};
 288         $self->{stream_state} = DISCONNECTED;
 289     }
 290 }
 291
 292 sub characters {
 293     my($parser, $chars) = @_;
 294     my $self = $parser->{_parent_};
 295     my $state = $self->{xml_state};
 296
 297     if($state == IN_BODY) {
 298         $self->{message}->{body} .= $chars;
 299
 300     } elsif($state == IN_THREAD) {
 301         $self->{message}->{thread} .= $chars;
 302
 303     } elsif($state == IN_STATUS) {
 304         $self->{message}->{status} .= $chars;
 305     }
 306 }
 307
 308 sub end_element {
 309     my($parser, $name) = @_;
 310     my $self = $parser->{_parent_};
 311     $self->{xml_state} = IN_NOTHING;
 312
 313     if($name eq 'message') {
 314         $self->push_msg($self->{message});
 315         $self->{message} = OpenSRF::Transport::SlimJabber::XMPPMessage->new;
 316
 317     } elsif($name eq 'stream:stream') {
 318         $self->{stream_state} = DISCONNECTED;
 319     }
 320 }
 321
 322 sub flush_socket {
 323         my $self = shift;
 324         my $socket = $self->socket;
 325     return 0 unless $socket and $socket->connected;
 326
 327     my $flags = fcntl($socket, F_GETFL, 0);
 328     fcntl($socket, F_SETFL, $flags | O_NONBLOCK);
 329
 330     while( my $n = sysread( $socket, my $buf, 8192 ) ) {
 331         $logger->debug("flush_socket dropped $n bytes of data");
 332         $logger->error("flush_socket dropped data on disconnected socket: $buf")
 333             unless($socket->connected);
 334     }
 335
 336     fcntl($socket, F_SETFL, $flags);
 337     return 0 unless $socket->connected;
 338     return 1;
 339 }
 340
 341
 342
 343
 344
 345 1;
 346
 347
 348
 349
 350