rest.pm

Punjab is a jabber XMLRPC/SOAP/REST client. It is a xmlrpc, soap, or REST server that allows for p

package PunJab::Server::REST;
use strict;
use warnings;

use POE;
use POE::Filter::XML;
use POE::Filter::XML::Node;
use POE::Filter::XML::NS qw/ :JABBER :IQ /;
use POE::Filter::XML::Utils;
use POE::Filter::Stream;

use HTTP::Status;

use PunJab::Util;
use PunJab::Session;
use PunJab::Server::HTTP;

use POSIX qw(strftime);
use Digest::SHA1;
use Digest::MD5 qw(md5 md5_hex md5_base64);
use Data::Uniqid qw ( suniqid uniqid luniqid );
use URI::QueryParam;


my %public_interfaces;


sub new {
    shift;
    my $config = shift;
    my $alias  = shift;
    $alias = "rest" if not defined $alias;

    PunJab::Server::HTTP->new($config,\&handler,\&streamer,\&rest_error,'rest_web_server');
       
    POE::Session->create
	( inline_states =>
	  { _start => sub {
	      my $kernel = $_[KERNEL];

	      if ($config->{debug}>0) {
		  &debug("REST Server Session Started\n");
	      }
	      $kernel->alias_set($alias);

	  },
	  publish => sub {
	      my ($alias, $event) = @_[ARG0, ARG1];
	      
	      $public_interfaces{$alias}{$event} = 1;
	      
	  },
	  rescind => sub {
	      my ($alias, $event) = @_[ARG0, ARG1];
	      delete $public_interfaces{$alias}{$event};
	  },
	#    send_rest     => \&send_rest,
         },
	  heap => { CONFIG => $config },
	  options => { debug => $config->{'debug'}, trace => $config->{'debug'} }
	  );
    
    return undef;

}


### Handlers for the rest server. 
sub shutdown_service {
    # loop and delete all interfaces
    
    $_[KERNEL]->yield("shutdown");
}

sub rest_error()
{
     my ($kernel, $heap, $sender, $syscall_name, $error_number, $error_string) = 
	 @_[KERNEL, HEAP, SENDER, ARG0 .. ARG2];
     my $params;
     # when client dies we need to kill the stream event
     
     &debug($syscall_name);
     &debug($error_number);
     &debug($error_string);
}

sub streamer()
{
     my ($kernel, $heap, $sender) = @_[KERNEL, HEAP, SENDER];
     my $params;
     
     if (defined $heap->{'client'}->[2]->{'header'} and ref $heap->{'client'}->[2]->{'header'} eq "HTTP::Request") {
	      
	 for my $key ($heap->{'client'}->[2]->{'header'}->uri->query_param) {
	     $params->{$key}= $heap->{'client'}->[2]->{'header'}->uri->query_param($key);
	 }
     
     }
     
     if (defined $params->{'stream'} and $params->{'stream'} == 1) {
	 $heap->{client}->set_output_filter( POE::Filter::Stream->new() );
	 
	 if ($heap->{'buffer'} ne "") {
	     $heap->{'last_stream_count'} = $heap->{'stream_count'};
	     #$heap->{'stream_count'}++;
	 }
	 &debug(" last ($heap->{'last_stream_count'})  now : $heap->{'stream_count'} \n\n");
	 	 
	 my @method       = split(/\//,$heap->{'client'}->[2]->{'header'}->uri->path());
	 shift @method;
	 my $method_name = shift @method;

	 if ($method_name eq "messages" or $method_name eq "iqs" or $method_name eq "presences") {
	     if ($heap->{'stream_count'}==0) {
		 $heap->{'client'}->put("<stream>");
		 $heap->{'stream_count'}++;
	     }	     
	     $heap->{'client'}->put($heap->{'buffer'});
	     $heap->{'buffer'} = "" if $heap->{'buffer'} ne "";
	     if ($heap->{'last_stream_count'} == $heap->{'stream_count'}) {
		 $heap->{'client'}->put("\n");
		 $heap->{'stream_count'}++;
	     }
	 } else {
	     $heap->{'client'}->put(qq {<?xml version="1.0"?>
<error>
<code>500</code>
<fault>Error in stream</fault>
<description>You can not stream this method!</description>
 </error>});

	     $kernel->yield('shutdown');
	 }
     }
     

}

sub handler()
{
    my ($kernel,$heap,$sender,$request) = @_[KERNEL, HEAP, SENDER, ARG0];
    if($request->isa('HTTP::Response'))
    {
         _request_failed( $request,
                          403,
                          "Bad Request",
                          "This was a response?",
			  $sender
                          );	 
         return;
     }
    $heap->{'stream_count'} = 0;

    my $response = HTTP::Response->new();
    my $query_string = $request->uri->query();
    
    my @requests     = split(/\|/,$request->uri->path()); # work on pipes later
    my @method       = split(/\//,$request->uri->path());
    shift @method;

    my $session                = 'rest_shell';
    my $http_method            = $request->method();
    my $request_content_type   = $request->header('Content-Type');
    my $request_content_length = $request->header('Content-Length');
    my $debug_request          = $request->header('DebugRequest');
    my $request_content        = $request->content();
    
    # parse request_content
    my $stdin                  = $request_content;
    my $method_name            = shift @method;
    my $args                   = \@method;


    unless (defined($method_name) and length($method_name)) {
	_request_failed( $response,
			 404,
			 "Missing Resource",
		         "You did not specify a tool. Example : http://localhost/help ",
			 $sender
			 );
	return;
    }
    
    unless ($method_name =~ /^(\S+)$/) {
	_request_failed( $response,
			 404,
			 "Bad Request",
			 "Unrecognized tool: $method_name",
			 $sender,
			 );
	return;
    }

    unless (exists $public_interfaces{$session}) {
	_request_failed( $response,
			 500,
			 "Bad Request",
			 "$session does not exist.",
			 $sender,
			 );
	return;
    }
    
    unless (exists $public_interfaces{$session}{$method_name}) {
	# put rest routing code here. 
	_request_failed( $response,
			 500,
			 "Bad Request",
			 "Unknown method: $method_name",
			 $sender,
                   );
	return;
    }
    $request->{'remote_ip'} = $heap->{'remote_ip'};
    eval {
	Transaction->start($sender,
			   $request, 
			   $response, 
			   $session, 
			   $method_name, 
			   $http_method, 
			   $args, 
			   $stdin);
    };
    
    if ($@) {
	_request_failed( $response,
			 500,
			 "Application Faulted",
			 "An exception fired while processing the request: $@",
			 $sender,
			 );
    }
    
}




sub _request_failed()
 {
  my ($response, $fault_code, $fault_string, $result_description,$sender) = @_;

  my $response_content = qq{<?xml version="1.0"?>
<error>
<code>$fault_code</code>
<fault>$fault_string</fault>
<description>$result_description</description>
</error>};
  $response->code($fault_code);
  $response->header("Content-Type", "text/xml");
  $response->header("Content-Length", length($response_content));
  $response->content($response_content);
  my $pb = $sender->postback( 'send_response', $response);
  $pb->($response);
}

package Transaction;
use URI::Split qw(uri_split uri_join);

sub TR_REQUEST  () { 0 }
sub TR_RESPONSE () { 1 }
sub TR_SESSION  () { 2 }
sub TR_EVENT    () { 3 }
sub TR_METHOD   () { 4 }
sub TR_ARGS     () { 5 }
sub TR_STDIN    () { 6 }
sub TR_PARAMS   () { 7 }
sub TR_CLIENT   () { 8 }

sub start {
  my ($type,$sender, $request, $response, $session, $event, $method, $args, $stdin) = @_;

  my $params;
  my $client; 

  for my $key ($request->uri->query_param) {
      $params->{$key}= $request->uri->query_param($key);
  }

  $stdin =~ s/\x00//gi; # get rid of crap

  my $self = bless
    [ $request,
      $response,
      $session,
      $event,
      $method,
      $args,
      $stdin,
      $params,
      $sender->postback( 'send_response', $response),
    ], $type;

  $POE::Kernel::poe_kernel->post($session, $event, $self);
  undef;
}

sub request {
    my $self = shift;
  return $self->[TR_REQUEST];
}


sub response {
    my $self = shift;
  return $self->[TR_RESPONSE];
}

sub params {
  my $self = shift;
  return $self->[TR_PARAMS];
}

sub args {
  my $self = shift;
  return $self->[TR_ARGS];
}

sub stdin {
  my $self = shift;
  return $self->[TR_STDIN];
}

sub uri {
  my $self = shift;
  my $noq  = shift;
  my $uri;
  if (defined($noq)) {
      my ($scheme, $auth, $path, $query, $frag) = uri_split($self->[TR_REQUEST]->uri);
      $uri = uri_join($scheme, $auth, $path, undef, $frag);
  } else {
      $uri = $self->[TR_REQUEST]->uri->as_string;
  }

  return $uri;
}


sub method {
  my $self = shift;
  return $self->[TR_METHOD];
}

sub return {
    my $self      = shift;
    my $output;
    my ($content) = shift;
    my $response  = $self->[TR_RESPONSE];
    my $params    = $self->[TR_PARAMS];

    # need to re-write this server

    if (ref($content) eq "POE::Filter::XML::Node") {
	$output = $content->to_str();
    } else {
	# this needs to be fixed
	$output = "<login><sid>".$content."</sid></login>";
    }
    $response->code(200);
    $output = "<?xml version=\"1.0\"?>\n" . $output;	
        
    if (!$params->{'stream'}) {
	
	$response->header("Content-Type", "text/xml");
	$response->header("Content-Length", length($output));
	$response->content($output);
    } else {
	$response->push_header("Content-Type", "text/xml");
    }
    $self->[TR_CLIENT]->($response,$params->{'stream'},$output);
    return;

}




sub auth {
    my $self = shift;
    my $output;
    my ($content) = shift;
    my $response = $self->[TR_RESPONSE];
    
    if (ref($content) eq "POE::Filter::XML::Node") {
	$output = $content->to_str();
    } else {
	# this needs to be fixed.
	$output = "<error><code>401</code><fault>Bad Authentication</fault><description>".$content."</description></error>";
    }

    $response->code(200);
    #$response->header("status", "401 Unauthorized");
    #$response->www_authenticate();
    #$response->header("WWW-authenticate", "basic realm=\"PunJab\"");
    $response->header("Content-Type", "text/xml");
    $response->header("Content-Length", length("<?xml version=\"1.0\"?>\n".$output));
    $response->content("<?xml version=\"1.0\"?>\n".$output);
    $self->[TR_CLIENT]->($response);
    return;

}

sub error {
    my $self = shift;
    my $content = shift;
    my $response = $self->[TR_RESPONSE];
    
    my $response_content = qq{<?xml version="1.0"?>
				  <error>
				  <code>500</code>
				  <fault>Application Faulted</fault>
				  <description>$content</description>
				  </error>};

    $response->code(500);
    $response->header("Content-Type", "text/xml");
    $response->header("Content-Length", length($response_content));
    $response->content($response_content);

    $self->[TR_CLIENT]->($response);
    return;
  
}

1;

__END__