Purple::_connection_error should also call the Component's connection_cb.

[thrasher.git] / perl / lib / Thrasher / Component.pm

package Thrasher::Component;
use strict;
use warnings;

=pod

=head1 NAME

Thrasher::Component - tie together XML stream processing and component
state into one module to handle a single "component"

=head1 DESCRIPTION

Thrasher::Component implements a XEP-0100 compliant component, with
hooks for adding further stuff into the component in a defined
way. The component is primarily "private" and documented in comments,
but this POD will document two things: The interface for the
Thrasher::Protocol implementations, and the hooks provided
for extending the base XEP-0100 protocol.

=cut

# General overview of this file: Some support code for the component
# is written in the front, then after the 
###### PROTOCOL SUPPORT
# comment below, we'll be implementing the component protocol 
# by following along the specification at
#  http://www.xmpp.org/extensions/xep-0114.html
# which will be referenced by section number.

# This file should only implement the bare minimum protocol stuff;
# additional capabilities should live elsewhere, just to avoid this
# file getting pointlessly large, and to avoid mixing extensions
# in with the XEP-0100 stuff.

# Right now this only permits one protocol per connection, but
# I'm trying to structure this for multi-protocol support in the
# future, per the later XEP on components.

use Thrasher::Log qw(:all);
use Thrasher::XMPPStreamOut;
use Thrasher::XMPPStreamIn qw(:all);
use Thrasher::Constants qw(:all);
use Thrasher::Plugin qw(:all);
use Thrasher::XML qw(:all);
use Thrasher::ConnectionManager qw(:all);
use Thrasher::XHTML_IM_Normalize qw(xhtml_and_text text);
use Thrasher::Callbacks qw(:all);

use Thrasher::Plugin::Basic;
use Encode;

use Carp qw(confess longmess);
use Data::Dumper;

use Digest::SHA1 qw(sha1_hex);

use base 'Exporter';

our @EXPORT_OK = qw(feature
                    has_subtags strip_resource no_match);
our %EXPORT_TAGS = (all => \@EXPORT_OK);

my $DEBUG = $Thrasher::DEBUG;

# This indicates whether or not to use the connection manager to
# prevent flooding and to tell whether or not the remote service is
# having trouble. This should generally only be turned off for 
# debugging purposes, but it could be useful in other scenarios.
our $USE_CONNECTION_MANAGER = 1;

# For testing purposes, this allows use to simply tell the component
# whether it is getting directly connected or not, so we can
# test the XML output it is supposed to generate. In real execution
# this should never be set.
our $WILL_BE_DIRECTLY_CONNECTED = undef;

# Hmm, this isn't good, but ProxyFileTransfer needs it. When/if
# we ever want to run multiple components out of one protocol we
# need to fix this.
our $COMPONENT;

# This manages the input from and the output to the component, but is
# not responsible for managing the socket; it receives XML text and
# is expected to output XML text. This, again, allows us to trivially
# test this component in a unit test environment without having to
# connect to a real XMPP server.

# Text is received by this object by calling "->text_in". Text out
# is sent out along the closure received during object construction,
# converted through Thrasher::XMPPStreamOut into text. A 
# method ->xml_out accepts XML for output, but should be internal-only.

# used to make unique IDs
our $id = 1;
sub get_id {
    return "id" . ($id++);
}

# States change mostly during connection. This changes which 
# functions are passed the incoming XML events.
my $states = {
    'disconnected' => sub { },
    connecting => \&xml_in_connecting,
    handshaking => \&xml_in_handshaking,
    connected => \&xml_in_connected
};

sub new {
    my $class = shift;
    my $self = {};
    bless $self, $class;

    $self->{protocol} = shift;
    if (!UNIVERSAL::isa($self->{protocol}, 'Thrasher::Protocol')) {
        die "The first argument to the component needs to be "
            ."a Thrasher::Protocol instance.";
    }

    my $text_output_closure = shift;
    if (ref($text_output_closure) ne 'CODE') {
        die "Creating a component requires a closure for the output"
            ." of XML text.";
    }
    $self->{output} = $text_output_closure;

    $self->setup_streams;

    # Need info for the stream connect
    $self->{secret} = shift;
    $self->{component_name} = shift;

    $self->{state} = 'disconnected';
    $self->{xml_buf} = [];

    # This holds a jid => {registration_info => $registration_info,
    # error_xml => $xml_tag} hash. If we get a login request, but
    # we've already tried that registration info and the remote
    # service told us it's bad, we don't re-try it. This is
    # deliberately held in transient memory storage.
    $self->{authentication_errors} = {};

    $COMPONENT = $self;

    return $self;
}

sub setup_streams {
    my $self = shift;

    my $out_stream = new Thrasher::XMPPStreamOut($self->{output});
    $self->{out_stream} = $out_stream;

    my $in_stream = Thrasher::XMPPStreamIn::get_parser();
    $self->{in_stream} = $in_stream;
}

# The usual xml_out used for most traffic, will buffer the 
# XML if we're not currently connected.
sub xml_out {
    my $self = shift;

    if ($self->{state} ne 'connected') {
        push @{$self->{xml_buf}}, @_;
        return;
    }
    $self->{out_stream}->output(@_);
}

# xml_out used by connection routines, to force out the
# necessary handshaking.
sub force_xml_out {
    my $self = shift;
    $self->{out_stream}->output(@_);
}

sub output_initial_stream_tag {
    my $self = shift;
    
    my $initial_stream = 
        [[$NS_STREAM, 'stream'],
         {"{$NS_COMPONENT}to" => $self->{component_name}},
         []];
    $self->set_state('connecting');
    # This is a direct call to output_tag_and_children so we can
    # pass in the $is_root_element value, which this needs.
    $self->{out_stream}->output_tag_and_children($initial_stream, 1);
}

# Once we know we're connected, probe everybody in our list 
sub initialize_connection {
    my $self = shift;

    if ($self->{initialized}) {
        log("Already initialized connection.");
        return;
    }

    $self->{initialized} = 1;

    log("Initializing connection");

    my $backend = $self->{protocol}->{backend};
    my $all_jids = $backend->all_jids;

    for my $jid (@$all_jids) {
        $self->send_presence_xml($jid, 'probe');
    }

    # Send a discovery request at the server, which we
    # then recurse down one level to get the info for
    # those items.
    # But only do it if we have a SERVER NAME.
    if ($Thrasher::SERVER_NAME) {
      $self->iq_query
        ([[$NS_COMPONENT, 'iq'],
          {to => $Thrasher::SERVER_NAME,
           from => $self->{component_name},
           type => 'get'},
          [[[$NS_DISCO_ITEMS, 'query'], {}, []]]],
         sub {
             my $component = shift;
             my $iq_params = shift;
             my $iq_packet = shift;

             if ($iq_params->{type} eq 'error') {
                 # FIXME: Schedule a timeout to try again.
                 log("Server discovery failed, this may cause "
                     ."odd, random problems.");
                 failed("server_discovery_items");
                 return;
             }

             my $items = 
                 extract_disco_items($iq_params->{query});
             my %items_hash = map { $_ => 1} @$items;

             my $item_count = scalar(@$items);

             # For each item, fire off an info request
             for my $item (@$items) {
                 if ($item eq $self->{component_name}) {
                     $item_count--;
                     next;
                 }

                 $self->iq_query
                     ([[$NS_COMPONENT, 'iq'],
                       {to => $item,
                        from => $self->{component_name},
                        type => 'get'},
                       [[[$NS_DISCO_INFO, 'query'], {}, []]]],
                      sub {
                          my $component = shift;
                          my $iq_params = shift;
                          my $iq_packet = shift;
                          debug("Disco info got $iq_params->{type} from $item"
                                  . " with $item_count remaining.");

                          if ($iq_params->{type} ne 'error') {
                              my ($identities, $features) =
                                  extract_disco_info($iq_params->{query});
                              $Thrasher::SERVER_INFO->{$item} =
                                  [$identities, $features];
                          } else {
                              # Server was configured to return this
                              # existed, but it doesn't seem to.
                              # Hopefully the proxy service isn't
                              # transient
                              delete $Thrasher::SERVER_INFO->{$item};
                          }
                          
                          $item_count--;

                          if ($item_count == 0) {
                              succeeded("server_discovery_items");
                          }
                      });
                 # Ensure server_discovery_items callbacks fire
                 # eventually even when some component doesn't reply.
                 # That would be one evil component...
                 my $no_reply_check_timeout = sub {
                     # after 30s...
                     if ($item_count == scalar(@{$items})) {
                         # ...not a single component responded?!!
                         failed('server_discovery_items');
                     }
                     elsif ($item_count > 0) {
                         # ...got some responses, but not all.
                         # Give what we have to success callbacks--if
                         # we luck out the missing component(s) aren't
                         # the ones they need.
                         succeeded('server_discovery_items');
                     }
                     return 0; # never repeat.
                 };
                 $self->{'event_loop'}->schedule($no_reply_check_timeout,
                                                 30000);
             }
           });
    }
}

sub set_state {
    my ($self, $state) = @_;
    $self->{state} = $state;

    if ($state eq 'connected') {
        log("State set to 'connected'");
        $self->initialize_connection;
    }
}

sub xml_in {
    my $self = shift;
    my $xml = shift;
    
    my $state_xml_func = $states->{$self->{state}};
    if (!defined($state_xml_func)) {
        die "Receiving xml, but I don't have a handler for "
            ."state '" . $self->{state} . "', how odd! (1)";
    }
    
    my $parsed = $self->{in_stream}->parse($xml);
    for my $message (@$parsed) {
        $state_xml_func->($self, $message);
        
        # State may change after processing a message
        $state_xml_func = $states->{$self->{state}};
        if (!defined($state_xml_func)) {
            die "Receiving xml, but I don't have a handler for "
                ."state '" . $self->{state} . "', how odd! (2)";
        }
    }
}



######
## State handlers; mostly for connection, as once we're connected
## this component is basicly in a steady state. (The individual
## connections are more complicated, but not this.)
######

my $STREAM = [$NS_STREAM, 'stream'];
my $HANDSHAKE = [$NS_COMPONENT, 'handshake'];

# In this state, we've sent the original <stream:stream ...> tag,
# and we're expecting the stream tag from the server
sub xml_in_connecting {
    my $self = shift;
    my $xml_message = shift;

    eval {
        multi_extract(
            $xml_message,

            # Expected case - stream returned from the server.
            # Annoyingly, we can't really check for stream errors
            # at this level, since the stream tag is exactly the
            # same for success and failure, EXCEPT that we get
            # an additional error tag upon failure.
            [$STREAM, {from => save('host'), id => save('stream_id')}] =>
              sub {
                  # Server is on the track we expect, send out the
                  # handshake
                  my $params = shift;
                  my $handshake = lc(sha1_hex($params->{stream_id}
                                                .$self->{secret}));
                  $self->set_state('handshaking');

                  $self->force_xml_out([$HANDSHAKE, {}, [$handshake]]);
              });
    };
    if ($@) {
        log("Error in stream tag? Reconnecting:\n$@\n");
        $self->reconnect_stream();
    }
}

sub xml_in_handshaking {
    my $self = shift;
    my $xml_message = shift;

    # If this passes, we're connected.
    eval {
        multi_extract($xml_message,
                      [[$NS_COMPONENT, 'handshake'], {}, []] =>
                        sub {
                            $self->set_state('connected');
                            callbacks('connected', $self);
                        },

                      [[$NS_STREAM, 'error'], {}, save('text')] =>
                        sub {
                            my $params = shift;
                            die "Stream error after handshake. Server said: $params->{text}";
                        });
    };
    if ($@) {
        log("Handshake error; reconnecting:\n$@\n");
        $self->reconnect_stream();
    }
}

# In this state, we are connected, and are receiving arbitrary
# packets from arbitrary users.
sub xml_in_connected {
    my $self = shift;
    my $xml_message = shift;

    # Route the XML message according to the nature of the message.
    multi_extract($xml_message,

                  # IQ messages
                  [[$NS_COMPONENT, 'iq'],
                   {type => save("type"),
                    from => save("from"),
                    to => save("to"),
                    id => save("id"),
                    "{$NS_XML}lang" => save("language", 1)
                   },
                   # Save first child under "query" whether or not
                   # that's the actual tag name (e.g. "si").
                   save_match('query', [undef, undef, undef], 1)] =>
                  sub {
                      my $iq_params = shift;

                      # Get and set handlers
                      return $self->handle_iq($iq_params,
                                              $xml_message);
                  },

                  [[$NS_COMPONENT, 'presence'],
                   undef, undef] =>
                  sub {
                      callbacks('presence_in',
                                $self,
                                sub { $self->handle_presence($_[0]) },
                                $xml_message);
                  },

                  [[$NS_COMPONENT, 'message'], {
                      to => save('to'),
                      from => save('from'),
                      type => save('type', 1),
                   },
                   [
                       save_match('chatstate',
                                  [[$NS_CHATSTATES, undef], undef, undef],
                                  1),
                       save_match('body',
                                  [[undef, 'body'], undef, undef],
                                  1),
                   ]] =>
                  sub {
                      my $message_params = shift;
                      $message_params->{'type'} ||= 'chat';
                      if ($message_params->{'chatstate'}) {
                          $message_params->{'chatstate'}
                            = $message_params->{'chatstate'}->[0]->[1];
                      }
                      $self->handle_message($message_params->{to},
                                            $message_params->{from},
                                            $message_params->{body},
                                            $message_params->{'type'},
                                            $message_params->{'chatstate'});
                  },

                  # Stream error
                  [[$NS_STREAM, 'error']] =>
                  sub {
                      my $children = $xml_message->[2];
                      my $first_tag;
                      for my $child (@$children) {
                          if (ref($child) eq 'ARRAY') {
                              $first_tag = $child;
                              last;
                          }
                      }

                      if (!$first_tag) {
                          # No error tag? Shouldn't happen. Panic!
                          $self->terminate;
                          return;
                      }

                      my $tag_name = $first_tag->[0]->[1];
                      # If it's a "not well formed" error,
                      # we can try to reconnect. If it's anything
                      # else, panic.
                      if ($tag_name eq 'xml-not-well-formed') {
                          $self->reconnect_stream;
                      } else {
                          $self->terminate;
                      }
                  },

                  # Default handler - complain about the unknown
                  # packet, but otherwise ignore it.
                  undef() =>
                  sub {
                      log "Unexpected packet: " . Dumper($xml_message);
                  }
                  );
}

# $IQ_CALLBACKS{"${jid_without_resource}-${id}"} => \&callback;
#
# The request/response ID (generated in iq_query()) includes the bare
# JID in case another user tries to inject a forged response.
our %IQ_CALLBACKS;

sub handle_iq {
    my $self = shift;
    my $iq_params = shift;
    my $iq_packet = shift;

    my $id = $iq_packet->[1]->{'{}id'};

    my $request_id = strip_resource($iq_params->{'from'}) . '-' . $id;
    my $callback = $IQ_CALLBACKS{$request_id};
    if ($callback
          # Must not mistake unrelated requests using the same ID
          # scheme for the expected response (e.g. two Thrasher
          # instances doing server disco at the same time).
          && $iq_params->{type} =~ /^(?:result|error)$/) {
        local $@;
        eval {
            $callback->($self, $iq_params, $iq_packet);
        };
        log "IQ callback error: $@" if ($@);
        delete($IQ_CALLBACKS{$request_id});
        return;
    }

    if (! $iq_params->{query}
          || @{$iq_params->{query}} == 0
          || @{$iq_params->{query}->[0]} == 0) {
        # Unused and causes interesting issues when replying
        # especially if autovivification occurs.
        log('Skipping childless IQ: ' . Dumper($iq_packet));
        return;
    }

    my $query_ns = $iq_params->{query}->[0]->[0];
    my $query_type = $iq_packet->[1]->{'{}type'};
    my $target = 'client';
    if (!defined($iq_params->{to}) ||
        $iq_params->{to} eq $self->{component_name}) {
        $target = 'component';
    }
    my $func = method_for_iq_namespace($target, $query_type, $query_ns);

    # Allow ourselves to suppress the error for some namespaces.
    if ($func && $func eq 'ignore') {
        $self->iq_error($iq_params, 'service_unavailable');
        return;
    }

    if (!defined($func)) {
        log "Unexpected IQ query: " . Dumper($iq_params,
                                             $target, $query_type,
                                             $query_ns);
        if ($query_type ne 'error') {
            # Prevent loop with error response to error generating an error....
            $self->iq_error($iq_params, 'service_unavailable');
        }
        return;
    }

    return $func->($self, $iq_params, $iq_packet);
}

sub iq_query {
    my $self = shift;
    # Everything but the ID
    my $iq_packet = shift;
    my $callback = shift;

    my $id = get_id;
    $iq_packet->[1]->{id} = $id;

    if ($callback) {
        # get_id() never repeats within a Thrasher instance so the only way
        # we can get a duplicate ID in responses from the same bare JID
        # is if the user sends two from different resources. Oh, well.
        my $to = $iq_packet->[1]->{'to'}
          || $iq_packet->[1]->{'{}to'}
          || '';
        my $request_id = strip_resource($to) . '-' . $id;
        $IQ_CALLBACKS{$request_id} = $callback;
    }

    $self->xml_out($iq_packet);
}

##
###### PROTOCOL SUPPORT
##

sub send_presence_xml {
    my $self = shift;
    my $target_jid = shift;
    my $presence_type = shift;
    my $from_jid = shift || $self->{component_name};
    my $show = shift;
    my $status = shift;
    my $extra = shift;

    # target_jid can be unset when the presence tag is coming
    # from the transport itself
    if ($target_jid) {
        my $session = $self->session_for($target_jid);
        if ($session && $session->{status} eq 'disconnecting') {
            # Don't send presence info for connections we're 
            # currently disconnecting.
            log("Bypassing a presence from $from_jid because disconnecting");
            return; 
        }
    }

    {
        no warnings 'uninitialized';
        if ($target_jid eq $from_jid &&
            $target_jid eq $self->{component_name}) {
            log("Attempt to send presence to self: " . longmess);
        }

        if ($target_jid =~ /$self->{component_name}$/ &&
            $from_jid =~ /$self->{component_name}$/) {
            log("Attempting to send presence to self: "
                . Dumper($target_jid, $from_jid) . 
                "\n" . longmess);
        }
    }

    my @children;
    if ($show) {
        push @children, [[$NS_COMPONENT, 'show'], {}, [$show]];
    }
    if ($status) {
        push @children, [[$NS_COMPONENT, 'status'], {}, [$status]];
    }
    if ($extra) {
        push @children, @$extra;
    }

    my $presence_out_tag = [[$NS_COMPONENT, 'presence'],
                            {($presence_type ? (type => $presence_type) : ()),
                             from => $from_jid,
                             ($target_jid ? (to =>
                                             strip_resource($target_jid)) : ())},
                            \@children];

    callbacks('presence_out', 
              $self,
              sub { $self->xml_out($_[0]) },
              $presence_out_tag);
}

sub session_for {
    my $self = shift;
    my $session_for = shift;
    $session_for = strip_resource($session_for);
    return $self->{sessions}->{$session_for};
}

sub set_session_for {
    my ($self, $jid, $session) = @_;
    $jid = strip_resource($jid);

    $self->{'sessions'}->{$jid} = $session;
}

# Welcome to the ugliest function in all of Thrasher!
sub handle_presence {
    my $self = shift;
    my $presence_tag = shift;

    my ($element, $atts, $children) = @$presence_tag;

    for my $att qw(to from) {
        if (!$atts->{"{}$att"}) {
            log "Presence received with no '$att'; ignored.";
            return;
        }
    }

    my $type = $atts->{'{}type'};

    if ($type && $type eq 'error') {
        log("Got a presence error.");
        return;
    }

    # Section 4.1.1 #10 - our request accepted
    # FIXME: What if the request is rejected?
    if (defined($type) && 
        ($type eq 'subscribed' || $type eq 'unsubscribed') &&
        (!defined($atts->{'{}to'}) || 
         $atts->{'{}to'} eq $self->{component_name})) {
        return;
    }

    # Section 4.1.1 # 11
    if (defined($type) &&
        $type eq 'subscribe' &&
        $atts->{'{}to'} eq $self->{component_name}) {
        # Section 4.1.1 #12
        # Hey, sure, buddy, no problem
        # FIXME: There ought to be something about registration here.
        $self->xml_out([[$NS_COMPONENT, 'presence'],
                        {type => 'subscribed',
                         from => $self->{component_name},
                         to => $atts->{'{}from'}}, 
                        []]);
        return;
    }

    if (defined($type) &&
        $type eq 'unsubscribe' &&
        $atts->{'{}to'} eq $self->{component_name}) {
        # Section 4.3.1 #5
        # FIXME: Unregister?
        $self->xml_out([[$NS_COMPONENT, 'presence'],
                        {type => 'unsubscribed',
                         from => $self->{component_name},
                         to => $atts->{'{}from'}},
                        []]);
        return;
    }

    # Everything above here is there because it can be 
    # done without a session; below this, a session
    # is required

    my $from = strip_resource($atts->{'{}from'});
    my $session = $self->session_for($from);

    if (!defined($session) && $atts->{'{}type'}) {
        if ($atts->{'{}type'} && $atts->{'{}type'} eq 'probe') {
            # Not authorized.
            $self->send_presence_xml($atts->{'{}from'},
                                     'unavailable');
            return;
        }

        my $registration_info =
            $self->{protocol}->{backend}->registered($from);
        
        if (!defined($registration_info)) {
            if ($atts->{'{}from'} =~ /$self->{component_name}$/) {
                # Don't reply to what is effectively ourself.
                return;
            }

            if ($atts->{'{}to'} ne $self->{component_name}) {
                # If this was a directed presence and it wasn't 
                # directly for the transport, eat it.
                return;
            }

            $self->xml_out([[$NS_COMPONENT, 'presence'], 
                            {from => $self->{component_name},
                             to => $atts->{'{}from'},
                             type => 'error'},
                            [error_tag('registration_required')]]);
            return;
        } else {
            # A presence tag has been sent other than to log in,
            # such as to subscribe, but the user is not currently
            # logged in. If they are unsubscribing, go ahead
            # and say they are unsubscribed. Otherwise, this
            # is an error
            if ($atts->{'{}type'} eq 'unsubscribe') {
                $self->xml_out
                    ([[$NS_COMPONENT, 'presence'],
                      {from => $atts->{'{}to'},
                       to => $atts->{'{}from'},
                       type => 'unsubscribed'}, []]);
            } else {
                return;

                # This gets sent out after logging off; if I can
                # work out a way to distinguish that case vs. 
                # other cases where this would be called for, we can
                # put it back.
                $self->xml_out([[$NS_COMPONENT, 'presence'],
                                {from => $self->{component_name},
                                 to => $atts->{'{}from'},
                                 type => 'error'},
                                [error_tag('not_authorized')]]);
            }
            return;
        }
    }

    if (!defined($type)) {
        if (!defined($session)) {
            $self->login($atts->{'{}from'}, $presence_tag);
            if ($atts->{'{}to'} ne $self->{component_name}) {

            }
        } else {
            $self->echo_presence($session, $presence_tag);
        }
        return;
    }

    if ($type eq 'subscribe') {
        $session->subscribe($atts->{'{}to'});
        return;
    }

    if ($type eq 'unsubscribe') {
        $session->unsubscribe($atts->{'{}to'});
        return;
    }

    if ($type eq 'subscribed' || $type eq 'unsubscribed') {
        my $protocol = $self->{protocol};
        my $legacy_name =
            $self->xmpp_name_to_legacy($atts->{'{}from'},
                                       $atts->{'{}to'});
        if (!defined($legacy_name)) {
            log "No legacy name for " . $atts->{'{}to'};
            return;
        }
        $protocol->$type($session, $self, $legacy_name);

        return;
    }

    # FIXME: This conforms to the specification, but I think
    # we ought to track which resources are online and only
    # disconnect if the user has no resources online.
    if ($type eq 'unavailable') {
        if (!$atts->{'{}to'} ||
            $atts->{'{}to'} eq $self->{component_name}) {
            $self->logout($session);
            return;
        } else {
            # Maybe we should just skip this?
            log ("Presence unavailable not handled properly: "
                 .Dumper($atts));
            return;
        }
    }

    if (defined($session)) {
        $self->echo_presence($session, $presence_tag);
        return;
    }

    if ($type eq 'probe') {
        # We know who you are.
        $self->send_presence_xml($atts->{'{}from'}, '');
        return;
    }

    # This shouldn't be able to happen, all bases should be 
    # covered above.
    log "Received unexpected presence packet with no "
        . "associated session: \n" . Dumper($presence_tag);
}

# Echos the presence of the user back out to the protocol, be
# it a general update or a targetted update.
sub echo_presence {
    my $self = shift;
    my $session = shift;
    my $presence_tag = shift;

    my $type = $presence_tag->[1]->{'{}type'};
    my $show;
    my $status;
    for my $child (@{$presence_tag->[2]}) {
        if (ref($child) &&
            $child->[0]->[1] eq 'show') {
            $show = join '', @{$child->[2]};
        }
        if (ref($child) &&
            $child->[0]->[1] eq 'status') {
            $status = join '', @{$child->[2]};
        }
    }

    my $to = $presence_tag->[1]->{'{}to'};
    if ($to eq $self->{component_name}) {
        $self->{protocol}->user_presence_update
            ($session, $type, $show, $status);
    } else {
        my $target_user =
            $self->xmpp_name_to_legacy(strip_resource($presence_tag->[1]->{'{}from'}),
                                       $to);
        if ($target_user) {
            $self->{protocol}->user_targeted_presence_update
                ($session, $type, $show, $status, $target_user);
        } else {
            log "Sent targetted presence to user " 
                .$presence_tag->[1]->{'{}from'} . ", but I have no such user.";
        }
    }
}

sub login {
    my $self = shift;
    my $full_jid = shift;
    my $original_presence_tag = shift;
    my $jid = strip_resource($full_jid);

    # Already queued a past login attempt. Tell that attempt to use
    # the current full JID and don't enqueue another one.
    if ($self->{'connection_queued'}->{$jid}) {
        $self->{'connection_queued'}->{$jid} = $full_jid;
        $self->send_connection_queued($jid);
        return;
    }

    my $registration_info = $self->{protocol}->{backend}->registered($jid);

    if (my $error = $self->{authentication_errors}->{$jid}) {
        my $bad_registration_info = $error->{registration_info};
        if (compare_hashref($bad_registration_info,
                            $registration_info)) {
            # It looks like this only happens when users ask for it,
            # so dump out the XML.
            $self->xml_out($error->{error_xml});

            # Note there is one case this doesn't cover well; the user
            # entered the wrong password, gets it labelled as bad,
            # then actually CHANGES THE PASSWORD on the remote service
            # to match this password. I'll worry when that happens,
            # I guess, because right now the wrongness of pounding on
            # the remote service outweighs that chance.
            log("Discarding login attempt by $jid, because the "
                ."same registration info has already been labelled "
                ."as bad by the remote service.");
            return;
        }

        # It's a new registration, so try again. But first...
        delete $self->{authentication_errors}->{$jid};
    }

    my $login_handler = sub {
        # Result from session can be:
        # * ref (implies its the Session object)
        # * error string
        # * undef if there was an error and the Protocol is handling it.
        my $session_or_error = shift;

        # Failed login - Section 4.4.2
        if (!ref(my $error = $session_or_error)) {
            # Protocol can pass an error here to have XML generated or
            # roll its own.
            if ($error) {
                my $packet = [[$NS_COMPONENT, 'presence'],
                              {to => $full_jid,
                               from => $self->{component_name},
                               type => 'error'},
                              [error_tag($error)]];
                $self->xml_out($packet);
                if ($error eq 'not_acceptable') {
                    # Credential issue
                    $self->{authentication_errors}->{$jid} = {
                        registration_info => $registration_info,
                        error_xml => $packet,
                    };
                }
            }
            return;
        }

        # Success! Paranoia:
        delete $self->{authentication_errors}->{$jid};

        $self->send_presence_xml($full_jid, '');

        my $session = $session_or_error;
        # In case protocol didn't already associate the session.
        $self->set_session_for($jid, $session);
        $self->{protocol}->initial_login($session);
        if (defined($original_presence_tag) && 
            $original_presence_tag->[1]->{'{}to'} ne
                    $self->{component_name}) {
            $self->echo_presence($session, $original_presence_tag);
        }
    };

    if (!defined($registration_info)) {
        # FIXME: Determine if this happens and when; be sure
        # to check the possibility of us losing the registration
        # info while the user still thinks they are registered.
        log "$jid sent us available presence but has no "
            ."registration";
        return;
    }

    # Verify that we have all required components
    my @required_items = $self->{protocol}->registration_items;
    for my $item (@required_items) {
        if (!defined($registration_info->{$item})) {
            log("Registration item $item missing for $jid! Ack! Panic!");
            $self->{protocol}->{backend}->remove($jid);
            $self->xml_out([[$NS_COMPONENT, 'presence'],
                            {from => $self->{component_name},
                             to => $full_jid,
                             type => 'error'},
                            [error_tag('registration_required')]]);
            return;
        }
    }

    my $protocol_login = sub {
        if ($self->session_for($jid)) {
            # Racing? Can't call ->login() with a session already defined.
            log("login($full_jid) reached protocol_login"
                  . " but already has a session. WHAT IS GOING ON?\n");
            # Multiple active sessions for the same JID ends in tears.
            return;
        }

        # If the connect was queued, a more current resource may have
        # been stored since this closure was created.
        my $last_full_jid = delete($self->{'connection_queued'}->{$jid})
          || $full_jid;
        $self->{protocol}->login($login_handler,
                                 $registration_info,
                                 $last_full_jid,
                                 $self);
    };
    $self->{'connection_queued'}->{$jid} = $full_jid;

    if (!$USE_CONNECTION_MANAGER) {
        $protocol_login->();
    } else {
        my $immediate_connection = request_connect($protocol_login);
        if (defined($WILL_BE_DIRECTLY_CONNECTED)) {
            $immediate_connection = $WILL_BE_DIRECTLY_CONNECTED;
        }

        # In the event that an immediate connection is made, the rest
        # of the code already takes care of the presence tags.
        if (!$immediate_connection) {
            $self->send_connection_queued($jid);
        }
    }
}

# Fires our extended "connection queued" presence tag at the given bare JID.
sub send_connection_queued {
    my ($self, $jid) = @_;

    my $thrasher_presence =
      [[[$NS_THRASHER_PRESENCE, 'connection-queued'], {}, []]];
    $self->send_presence_xml($jid,
                             'unavailable',
                             undef,
                             undef,
                             'connection queued',
                             $thrasher_presence);
}

sub logout {
    my $self = shift;
    my $session = shift;
    my $extra = shift;
    my $logout_status_message = shift;

    # Accept JIDs for the session
    if (!ref($session)) {
        $session = $self->session_for($session);
    }

    if ($session->{status} =~ /disconnecting/) {
        log("Already logging out $session->{jid}, but got another "
            ."request to log out.");
        return;
    }

    log("Logging out $session->{jid}");

    $session->{status} = 'disconnecting before presence';

    my $logout_handler = sub {
        # One way or another, logging off is successful.
        # Send logout packets; this should show everybody who
        # isn't offline as offline.

        if (!defined($session)) {
            # FIXME: This shouldn't happen. 
            # Sequence to trigger:
            # * Register.
            # * kill transport, whack database.
            # * come online still subscribed, go offline.
            # * this is reached.
            # As you can guess, the "whack database" step is frankly
            # more hostility than we can really plan for.
            return;
            confess "Made it to logout handler without session.";
        }

        $session->logout($self);

        my $roster = $self->{protocol}->{backend}->get_roster
            ($session->{jid});

        my @on_roster = map {
            $self->{protocol}->{backend}->legacy_name_to_jid($session->{jid}, 
                                                             $_, 
                                                             $self->{component_name})
            } keys %$roster;
        for my $roster_entry (@on_roster) {
            $self->send_presence_xml($session->{jid}, 'unavailable',
                                     $roster_entry);
        }
        
        if ($extra && ref($extra) ne 'CODE') {
            log("Got 'extra' that isn't code: " . longmess);
        } elsif ($extra) {
            $extra->();
        }

        $self->send_presence_xml($session->{jid}, 'unavailable',
                                 undef, undef, $logout_status_message);

        $session->{status} = 'disconnecting';

        delete $self->{sessions}->{$session->{jid}};

        log("session disconnected for $session->{jid}");
    };

    # Do we also need to show all transport users as offline,
    # or does something in the server take care of that.
    $self->{protocol}->logout($session, $logout_handler);
}

sub handle_message {
    my $self = shift;
    my $to = shift;
    my $from = shift;
    my $body_xml = shift;
    my $type = shift;
    my $chatstate = shift;

    if (defined($type) && $type eq 'error') {
        log("Got an error message from a user.");
        return;
    }

    my $session = $self->session_for($from);

    # FIXME: There can be a race condition where the error sub is 
    # called after the user disconnects. We shouldn't send this then.
    my $error_handler = sub {
        my $error = shift;
        my $message = [[$NS_COMPONENT, 'message'],
                       {to => $from,
                        from => $to,
                        type => 'error'},
                       [
                        # $body_xml, # FIXME - better to send this?
                        error_tag($error)]];
        $self->xml_out($message);
    };

    my $error_message = [[$NS_COMPONENT, 'message'],
                         {to => $from,
                          from => $to,
                          type => 'error'},
                         [error_tag('registration_required'),
                          $body_xml]];
    if (!$session) {
        my $registration_info = $self->registration_info($from);

        if (!defined($registration_info)) {
            $self->xml_out($error_message);
            return;
        }

        # If we get here, the user has registered, and is sending
        # a message, but they are apparently not actually logged 
        # in, perhaps because they deliberately logged off. I'm
        # choosing to allow them to log in this way.
        # FIXME: Hey, actually do that. For now you get an error.
        $self->send_error_message
            ($from, "You must be logged in to send messages "
             ."to the remote service users.", 'service_unavailable',
             $to);
        return;
        
    } elsif (!$session->is_registered) {
        $self->xml_out($error_message);
        return;
    }

    my $converted_to = $self->xmpp_name_to_legacy($session->{jid},
                                                  strip_resource($to));

    # Tie successful call to the protocol to the successful extraction
    # of the message from the input
    if ($body_xml) {
        eval {
            my $body = extract([undef, undef,
                                save_sub("text", \&text_extractor)],
                               $body_xml);

            my $body_text = join '', @{$body->{text} || []};

            $self->{protocol}->send_message($session,
                                            $converted_to,
                                            $body_text,
                                            $type,
                                            $error_handler);
        };
        if ($@) {
            log("Error in extracting message from "
                  . Dumper($body_xml) . ":\n" . $@);
        }
    }

    if ($chatstate) {
        eval {
            $self->{protocol}->outgoing_chatstate($session,
                                                  $converted_to,
                                                  $chatstate);
        };
        if ($@) {
            log("Error in outgoing_chatstate:\n$@");
        }
    }

    if (! ($body_xml || $chatstate)) {
        log('Message without usable child.');
        return;
    }
}

sub registration_info {
    my $self = shift;
    my $jid = shift;
    my $stripped_jid = strip_resource($jid);
    return $self->{protocol}->{backend}->registered($stripped_jid);
}

sub legacy_name_to_xmpp {
    my $self = shift;
    my $user_jid = strip_resource(shift());
    my $legacy_name = shift;
    my $lang = shift || 'en';

    # FIXME: XMPP is correct
    # FIXME: Lang on the user
    return $self->{protocol}->{backend}->legacy_name_to_jid
        ($user_jid, $legacy_name, $self->{component_name}, $lang);
}

sub xmpp_name_to_legacy {
    my $self = shift;
    my $user_jid = strip_resource(shift());
    my $target_jid = strip_resource(shift());

    return $self->{protocol}->{backend}->jid_to_legacy_name
        ($user_jid, $target_jid);
}

=pod

=head1 PROTOCOL INTERFACE

The protocol interface is intended to sheild Thrasher::Protocol
implementers from potential changes to the Component interface.
If you, as a Thrasher::Protocol implementer ever feel compelled
to reach into the ::Component to do anything not accessible from
this interface, please let us know so we can give you a more 
official path.

The officially-implemented methods are:

=over 4

=item *
C<add_contact>($jid, $legacy_user_name): This will send out
the correct <presence> tag to attempt to add the $legacy_user_name
to the given $jid. This corresponds with section 5.1 in the XEP.

You should be able to retrieve the $jid out of the information you
stored in the ::Session, and you should send in the $legacy_user_name
as the raw username from the service; ::Component will take care of
mapping it as appropriate, in accordance with the name translation
protocols.

=cut

sub add_contact {
    my $self = shift;
    my $jid = shift;
    my $legacy_user_name = shift;

    my $legacy_jid = $self->legacy_name_to_xmpp
        ($jid, $legacy_user_name);

    $self->send_presence_xml($jid, 'subscribe', $legacy_jid);
}

=pod

=item *
C<send_presence>($jid, $legacy_user_name, $type, $show):
Send the given presence for the given legacy_user_name.
The ::Protocol implementation will need to convert the status
into an XMPP-status and give us the "type" and "show".

=cut

sub send_presence {
    my $self = shift;
    my $jid = shift;
    my $legacy_user_name = shift;
    my $type = shift;
    my $show = shift;
    my $status = shift;

    my $from_jid = $self->legacy_name_to_xmpp($jid, $legacy_user_name);

    if ($status) {
        $status = text($status);
    }

    my $session = $self->session_for($jid);
    $session->{component}->{presence}->{strip_resource($jid)}->{strip_resource($from_jid)} =
        [$type, $show, $status];

    $self->send_presence_xml($jid, $type, $from_jid, $show, $status);
}

=pod

=item *
C<delete_contact>($jid, $legacy_user_name): This will send out
the necessary packets to indicate that a user has unsubscribed.

=cut

sub delete_contact {
    my $self = shift;
    my $jid = shift;
    my $legacy_user_name = shift;

    my $legacy_jid = $self->legacy_name_to_xmpp($jid, $legacy_user_name);

    $self->send_presence_xml($jid, 'unsubscribe', $legacy_jid);
    $self->send_presence_xml($jid, 'unsubscribed', $legacy_jid);
    # FIXME: Example 50 says this should be to the JID w/ 
    # the resource
    $self->send_presence_xml($jid, 'unavailable', $legacy_jid);
}

=pod

=item *
C<send_message>($jid_from, $jid_to, $message, $extra): Sends a message
from the given jid to the given jid. $extra is a hash containing extra
parametrs, which include:

=over 4

=item *

C<$is_xhtml_ish>: If false, sends the UTF-8 encoded $message to the
target $jid_to.

If it is true, it will process the XHTML-ish message into an
XHTML and a plain text string, and send the XHTML-ish message
as an XHTML-IM message in complaince with XEP-0071. Note that
there is a normalization step, so you don't need to sweat
whether it is proper XHTML; this does a decent job of turning 
dreck into XHTML.

=item *

C<$nick>: If set to a true string, will broadcast the nick conforming
to XEP-0172. Note that according to the XEP, nickname should be
broadcast only once per connection per (legacy) user, and it 
is your responsibility to ensure this, not this method's.

=item *

C<$type>: If set, will set the type of the message to the given 
XMPP type.

=back

=cut

sub send_message {
    my $self = shift;
    my $jid_from = shift;
    my $jid_to = shift;
    my $message = shift;
    my $extra = shift;

    my $type = $extra->{type} || 'chat';
    my $is_xhtml_ish = $extra->{is_xhtml_ish};
    my $nick = $extra->{nick};
    my $extra_children = $extra->{children} || [];

    if ($nick) {
        $nick = [[[$NS_NICK, 'nick'], {}, [$nick]]];
    } else {
        $nick = [];
    }

    if ($jid_from =~ / / ||
        $jid_to =~ / /) {
        log("Trying to send/receive message from a JID with "
            ."a space in it: from: $jid_from to: $jid_to "
            ."\n" . longmess);
        return;
    }

    if (!$is_xhtml_ish) {
        $self->xml_out([[$NS_COMPONENT, 'message'],
                        {from => $jid_from,
                         to => $jid_to,
                         type => $type},
                        [[[$NS_COMPONENT, 'body'],
                          {},
                          [$message]],
                         @$nick, @$extra_children]]);
    } else {
        my ($xhtml, $text) = xhtml_and_text($message);
        $self->xml_out([[$NS_COMPONENT, 'message'],
                        {from => $jid_from,
                         to => $jid_to,
                         type => $type},
                        [[[$NS_COMPONENT, 'body'],
                          {}, [$text]],
                         @$nick, @$extra_children,
                         [[$NS_XHTML_IM, 'html'], {},
                          [[[$NS_XHTML, 'body'], {}, [\$xhtml]]]]]]);
    }
}

=pod

=item *

C<send_error_message>($jid, $error_msg): Sends an error message
to the user, coming from the transport.

In my experience, this should be limited, because this gets very
annoying very quickly. As the method name implies, reserve it
for errors.

You're responsible for providing the errors. The $session for a user
may have their language available to you in $session->get_lang,
but it depends on their XMPP client (and how carefully we picked the
language out of the stream).

=cut

sub send_error_message {
    my $self = shift;
    my $target_jid = shift;
    my $error_message = shift;
    my $error_type = shift;
    my $from = shift || $self->{component_name};
    
    my $error_body = [];
    
    if ($error_type) {
        push @$error_body, error_tag($error_type);
    }

    $self->send_message($from, $target_jid,
                        $error_message, 
                        {type => 'error', children => $error_body});
}

=pod 

=item *

C<set_roster_name>($jid, $legacy_jid, $name): Sets $jid's
roster entry to $legacy_jid to have the given nickname,
if $jid's client advertises support for XEP-0144, by 
sending a modify request.

=cut

sub set_roster_name {
    my $self = shift;
    my $jid = shift;
    my $legacy_jid = shift;
    my $name = shift;
    my $force = shift;

    my $session = $self->session_for($jid);

    my $send_iq = sub {
        my $iq = [[$NS_COMPONENT, 'iq'], 
                  {from => $self->{component_name},
                   to => $jid,
                   type => 'set'},
                  [[[$NS_ROSTER_EXCHANGE, 'x'], {},
                    [[[$NS_ROSTER_EXCHANGE, 'item'],
                      {action => 'modify',
                       jid => $legacy_jid,
                       name => $name}, []
                      ]]]]];
        $self->iq_query($iq);
    };

    if ($force) {
        $send_iq->();
    } else {
        $session->do_if_feature($NS_ROSTER_EXCHANGE, 
                                $send_iq);
    }
}
    
=pod

=back

=cut

# For some reason, we can no longer continue. Send all presence
# closing, terminate the connection, and terminate the mainloop.
sub terminate {
    my $self = shift;
    my %args = @_;

    $args{reason} ||= 'Internal error';

    if ($self->{I_AM_TERMINATING}) {
        return;
    }

    log("Component terminating");
    $self->{I_AM_TERMINATING} = 1;

    my $protocol = $self->{protocol};
    my $sessions = $self->{sessions};

    # If we are terminating because the DB lost the connection,
    # we no longer know enough to actually log people off. If
    # we are terminating due to a signal, or most other reasons,
    # we can log people off cleanly.
    if (!$args{no_db}) {
        for my $session (values %$sessions) {
            log("Terminating connection for $session->{jid}");
            $self->logout($session, undef, $args{reason});
        }
    }

    # And terminate the event loop, which is currently
    # hard-coded
    $self->{event_loop}->quit;
}

# This is for when the XMPP server stream simply disappears.
# This is probably because the server has crashed or gone down.
# In this case, we want the full terminate routine since it
# probably implies all users have been disconnected. 
# Unfortunately, we can't know this, but it's the best guess.
sub lost_connection {
    my $self = shift;

    $self->terminate;
}

# This is for when we have screwed up and borked our stream.
# If this ever triggers, it is almost certainly a bug in 
# Thrasher, but let's at least try to recover. We may lose
# some messages from the server in the meantime.
# FIXME: We ought to have a configuration setting for whether
# we try this recovery or just give up, because if you're
# using component load balancing, this will really screw your
# users up.
sub reconnect_stream {
    my $self = shift;

    log("Attempting to reconnect stream.");
    
    # This causes any events that may be generated by the protocol
    # side while we are reconnecting to be buffered.
    $self->set_state('disconnected');

    # By the time this is getting called, the socket is entirely gone.
    log("Closing socket");
    $self->{thrasher_socket}->close();

    local $@;
    eval { $self->{thrasher_socket}->connect(); };
    if ($@) {
        # We can't seem to connect to the server. This should
        # never happen, so just panic.
        log("Connection to server could not be re-established.");
        $self->terminate;
        return;
    }

    log("Connection to server re-established. Handshaking.");

    $self->setup_streams;
    $self->{thrasher_socket}->establish_fd_watch;

    # Re-begin connection process
    $self->output_initial_stream_tag;
}

sub socket_in_closure {
    my $self = shift;
    my $socket = shift;

    my $closure = sub {
        my $got_data = 0;
        while (1) {
            my $val = eval { $socket->read(); };
            if ($@) {
                log("$@");
                $self->lost_connection();
                return 0;
            }
            elsif (! defined($val)) {
                last;
            }
            else {
                $got_data = 1;
                debug("IN: $val");
                eval {
                    $self->xml_in($val);
                };
                if ($@) {
                    # Terminate immediately after an unhandled error.
                    # Ugly, but better than leaving protocol-side online
                    # but component-side unreachable from the XMPP server
                    # because only the FD watch has gone.
                    log("Fatal error handling XML input:\n$@\n");
                    $self->terminate();
                    return 0;
                }
            }
        }
        if (!$got_data) {
            log "Connection to XMPP server lost.";
            $self->lost_connection();
            return 0;
        }
        return 1;
    };

    return $closure;
}

sub compare_hashref {
    my $a = shift;
    my $b = shift;

    if (scalar(keys %$a) != scalar(keys %$b)) {
        return 0;
    }

    while (my ($key, $value) = each %$a) {
        if ($b->{$key} ne $value) {
            return 0;
        }
    }

    return 1;
}

1;