Mail::Karmasphere::Client - Client for Karmasphere Reputation Server

 use Mail::Karmasphere::Client qw(:all);

 my $client = new Mail::Karmasphere::Client(
    PeerAddr	=> 'query.karmasphere.com',
    PeerPort	=> 8666,
    Principal   => "my_assigned_query_username",
    Credentials => "my_assigned_query_password",
    # see http://my.karmasphere.com/devzone/client/configuration#credentials
    # quickstart:  use temporary credentials for "generic perl".
    # recommended: use permanent credentials -- register for an account.
   );

 my $query = new Mail::Karmasphere::Query();
 $query->identity('127.0.0.2', IDT_IP4);
 $query->composite('karmasphere.email-sender');
 my $response = $client->ask($query, 6);
 print $response->as_string;

 my $id = $client->send($query);
 my $response = $client->recv($query, 12);
 my $response = $client->recv($id, 12);

 my $response = $client->query(
 	Identities	=> [ ... ]
 	Composite	=> 'karmasphere.email-sender',
 		);

package Mail::Karmasphere::Client;

use strict;
use warnings;
use vars qw($VERSION @ISA @EXPORT_OK %EXPORT_TAGS
				%QUEUE @QUEUE $QUEUE);
use Exporter;
use Data::Dumper;
use Convert::Bencode qw(bencode bdecode);
use IO::Socket::INET;
use Time::HiRes;
use IO::Select;
use Socket;
use constant {
	IDT_IP4_ADDRESS		=> "ip4",
	IDT_IP6_ADDRESS		=> "ip6",
	IDT_DOMAIN_NAME		=> "domain",
	IDT_EMAIL_ADDRESS	=> "email",

	IDT_IP4				=> "ip4",
	IDT_IP6				=> "ip6",
	IDT_DOMAIN			=> "domain",
	IDT_EMAIL			=> "email",
	IDT_URL				=> "url",
};
use constant {
	AUTHENTIC					=> "a",
	SMTP_CLIENT_IP				=> "smtp.client-ip",
	SMTP_ENV_HELO				=> "smtp.env.helo",
	SMTP_ENV_MAIL_FROM			=> "smtp.env.mail-from",
	SMTP_ENV_RCPT_TO			=> "smtp.env.rcpt-to",
	SMTP_HEADER_FROM_ADDRESS	=> "smtp.header.from.address",

	FL_FACTS		=> 1,
	FL_DATA			=> 2,
	FL_TRACE		=> 4,
	FL_MODELTRACE	=> 8,
};
use constant {
	PROTO_TCP		=> 0+getprotobyname('tcp'),
	PROTO_UDP		=> 0+getprotobyname('udp'),
};

BEGIN {
	@ISA = qw(Exporter);
	$VERSION = "2.18";
	@EXPORT_OK = qw(
					IDT_IP4_ADDRESS IDT_IP6_ADDRESS
					IDT_DOMAIN_NAME IDT_EMAIL_ADDRESS

					IDT_IP4 IDT_IP6
					IDT_DOMAIN IDT_EMAIL
					IDT_URL

					AUTHENTIC
					SMTP_CLIENT_IP
					SMTP_ENV_HELO SMTP_ENV_MAIL_FROM SMTP_ENV_RCPT_TO
					SMTP_HEADER_FROM_ADDRESS

					FL_FACTS
					FL_DATA
					FL_TRACE
					FL_MODELTRACE
				);
	%EXPORT_TAGS = (
		'all' => \@EXPORT_OK,
		'ALL' => \@EXPORT_OK,
	);
	%QUEUE = ();
	@QUEUE = ();
	$QUEUE = 100;
}

# We can't use these until we set up the above variables.
use Mail::Karmasphere::Query;
use Mail::Karmasphere::Response;

sub new {
	my $class = shift;
	my $self = ($#_ == 0) ? { %{ (shift) } } : { @_ };

	if ($self->{Debug} and ref($self->{Debug}) ne 'CODE') {
		$self->{Debug} = sub { print STDERR Dumper(@_); };
	}
	$self->{Debug}->('new', $self) if $self->{Debug};

	unless ($self->{Socket}) {
		$self->{Proto} = 'udp'
						unless defined $self->{Proto};
		$self->{PeerAddr} = $self->{PeerHost}
						unless defined $self->{PeerAddr};
		$self->{PeerAddr} = 'query.karmasphere.com'
						unless defined $self->{PeerAddr};
		$self->{PeerPort} = 8666
						unless $self->{Port};
		_connect($self);
	}

	return bless $self, $class;
}

sub _connect {
	my $self = shift;
	$self->{Debug}->('connect') if $self->{Debug};
	$self->{Socket} = new IO::Socket::INET(
		Proto			=> $self->{Proto},
		PeerAddr		=> $self->{PeerAddr},
		PeerPort		=> $self->{PeerPort},
		ReuseAddr		=> 1,
	);
	unless (defined $self->{Socket}) {
		delete $self->{Socket};
		my @args = map { "$_=" . $self->{$_} } keys %$self;
		die "Failed to create socket: $! (@args)";
	}
}

sub query {
	my $self = shift;
	return $self->ask(new Mail::Karmasphere::Query(@_));
}

sub _previous_socket {
	my $self = shift;
	return undef unless exists $self->{PreviousTime};
	if ($self->{PreviousTime} + 10 > time()) {
		$self->{Debug}->('previous') if $self->{Debug};
		return $self->{PreviousSocket}
	}
	delete $self->{PreviousSocket};
	delete $self->{PreviousTime};
	return undef;
}

sub _is_tcp {
	my ($self, $socket) = @_;
	return $socket->protocol == PROTO_TCP;
}

sub _send_real {
	my ($self, $data) = @_;

	my $socket = $self->{Socket};

	if ($socket->protocol == PROTO_UDP) {
		if (length $data > 1024) {	# Server's UDP_MAX
			$self->{Debug}->('fallback') if $self->{Debug};
			$self->{PreviousSocket} = $self->{Socket};
			$self->{PreviousTime} = time();
			$self->{Proto} = 'tcp';
			# XXX This loses the old socket and any queries
			# sent thereon.
			$self->_connect();
			$socket = $self->{Socket};
		}
	}
	# This can NOT be an else as we clobber the variable above.
	if ($socket->protocol == PROTO_TCP) {
		$self->{Debug}->('tcp prefix') if $self->{Debug};
		$data = pack("N", length($data)) . $data;
		$self->{Debug}->('send_real', $data) if $self->{Debug};
	}
	$socket->send($data)
					or die "Failed to send to socket: $!";
}

sub send {
	my ($self, $query) = @_;

	die "Not blessed reference: $query"
			unless ref($query) =~ /[a-z]/;
	die "Not a query: $query"
			unless $query->isa('Mail::Karmasphere::Query');

	$self->{Debug}->('send_query', $query) if $self->{Debug};

	my $id = $query->id;

	my $packet = {
		_	=> $id,
	};
	$packet->{i} = $query->identities if $query->has_identities;
	$packet->{s} = $query->composites if $query->has_composites;
	$packet->{f} = $query->feeds if $query->has_feeds;
	$packet->{c} = $query->combiners if $query->has_combiners;
	$packet->{fl} = $query->flags if $query->has_flags;
	if (defined $self->{Principal}) {
		my $creds = defined $self->{Credentials} ?$self->{Credentials} : '';
		$packet->{a} = [ $self->{Principal}, $creds ];
	}
	$self->{Debug}->('send_packet', $packet) if $self->{Debug};

	my $data = bencode($packet);
	$self->{Debug}->('send_data', $data) if $self->{Debug};

	$self->_send_real($data);

	return $id;
}

sub _recv_real {
	my ($self, $socket) = @_;

	my $data;
	if ($socket->protocol == PROTO_TCP) {
		$socket->read($data, 4)
					or die "Failed to receive length from socket: $!";
		my $length = unpack("N", $data);
		$data = '';
		while ($length > 0) {
			my $block;
			my $bytes = $socket->read($block, $length)
						or die "Failed to receive data from socket: $!";
			$data .= $block;
			$length -= $bytes;
		}
		$self->{Debug}->('recv_data', $data) if $self->{Debug};
	}
	else {
		$socket->recv($data, 8192)
					or die "Failed to receive from socket: $!";
		$self->{Debug}->('recv_data', $data) if $self->{Debug};
	}
	my $packet = bdecode($data);
	die $packet unless ref($packet) eq 'HASH';

	my $response = new Mail::Karmasphere::Response($packet);
	$self->{Debug}->('recv_response', $response) if $self->{Debug};
	return $response;
}

sub recv {
	my ($self, $query, $timeout) = @_;

	my $id = ref($query) ? $query->id : $query;
	if (defined($id)) {
		if ($QUEUE{$id}) {
			$self->{Debug}->('recv_find', $id, $QUEUE{$id})
							if $self->{Debug};
			@QUEUE = grep { $_ ne $id } @QUEUE;
			return delete $QUEUE{$id};
		}
	}
	else {
		if (@QUEUE) {
			$id = shift @QUEUE;
			return delete $QUEUE{$id};
		}
	}

	$timeout = 10 unless defined $timeout;
	my $finish = time() + $timeout;
	my $select = new IO::Select();
	$select->add($self->{Socket});
	my $prev = $self->_previous_socket;
	$select->add($prev) if $prev;
	while ($timeout > 0) {
		my @ready = $select->can_read($timeout);

		if (@ready) {
			my $response = $self->_recv_real($ready[0]);
			$response->{query} = $query if ref $query;
			return $response unless defined $id;
			return $response if $response->id eq $id;

			my $rid = $response->id;
			push(@QUEUE, $rid);
			$QUEUE{$rid} = $response;
			if (@QUEUE > $QUEUE) {
				my $oid = shift @QUEUE;
				delete $QUEUE{$oid};
			}
		}

		$timeout = $finish - time();
	}

	print STDERR "Failed to receive from socket: $!\n";
	return undef;
}

sub ask {
	my ($self, $query, $timeout) = @_;
	$timeout = 5 unless defined $timeout;
	for (0..2) {
		my $id = $self->send($query);
		my $response = $self->recv($query, $timeout);
		# $response->{query} = $query;
		return $response if $response;
		$timeout += $timeout;
	}
	return undef;
}

1;