File Coverage

blib/lib/POE/Component/Spread.pm

Criterion	Covered	Total	%
statement	13	15	86.6
branch			n/a
condition			n/a
subroutine	5	5	100.0
pod			n/a
total	18	20	90.0

package POE::Component::Spread;

use strict;
use vars qw($VERSION @ISA @EXPORT);
use Exporter;

$VERSION = "0.02";
@ISA = qw(Exporter);
@EXPORT = qw(REGULAR_MESS);

use Data::Dumper;
use POE::Driver::Spread;
use POE::Filter::Spread;
use Spread qw(:MESS);
use POE qw( Wheel::SocketFactory Wheel::ReadWrite Driver::SysRW Filter::Line Filter::Stream );

sub new {
    my( $package, $alias ) = splice @_, 0, 2;

    return POE::Session->create(
        package_states => [
            $package => [ qw(_get_spread _error _start publish subscribe connect disconnect) ]
        ],
        args => [ $alias, @_ ]
    );
}

sub _start {
    my ($kernel, $session, $heap, $alias) = @_[KERNEL, SESSION, HEAP, ARG0];
    $kernel->alias_set($alias);
}

sub connect {
    my $k_heap = $_[HEAP];   # kernel's heap
    my $sender = $_[SENDER];
    my $s_name = $_[ARG0];
    my $p_name = $_[ARG1];
    my $heap = $sender->get_heap();

    # Spread doesn't like hostnames without ports
    unless ($s_name =~ /^\d+$/ or $s_name =~ /@/) {
        $s_name = '4803@' . $s_name
    }

    my ($m, $pg) = Spread::connect( { private_name => $p_name, spread_name => $s_name } );
    print "m is $m, pg=$pg\n";

    if (!defined($m)) {
        $poe_kernel->post( $sender => '_error' => 'connect' );
    }

    $k_heap->{groups}->{$pg}->{$sender} = 'self';
    $heap->{private_name} = $pg;

    # create a filehandle from the fileno we get back from Spread::connect
    open $heap->{filehandle}, "<&=$m";

    $heap->{wheel} = POE::Wheel::ReadWrite->new(
            Handle => $heap->{filehandle},
            Driver => POE::Driver::Spread->new(mbox => $m),
            Filter => POE::Filter::Spread->new(),

            InputEvent => '_get_spread',
            ErrorEvent => '_error'
    );
    $heap->{spread} = $m;
}

sub disconnect {
    my $sender = $_[SENDER];
    my $heap = $sender->get_heap();

    $heap->{wheel}->shutdown_input();
    undef $heap->{wheel};
}

sub _error {
    my ($kernel, $session, $heap, $data, $sender, $id) = @_[KERNEL, SESSION, HEAP, ARG0, SENDER, ARG3];
    my $r_heap = $sender->get_heap();
#    print "ERROR: ID=$id session:[$session, $heap], sender:[$sender, $r_heap], kernel:$poe_kernel\n";
    # BLEH
}

sub _get_spread {
    my ($kernel, $session, $heap, $data) = @_[KERNEL, SESSION, HEAP, ARG0];
    my @moo = @$data;
    my ($type, $sender, $groups, $mess, $endian, $message) = @{$moo[0]};

    if (!defined($type)) {
        $kernel->post( $session => '_error' => 'null packet' );
        return undef;
    }

    foreach my $g (@$groups) {
        foreach my $r (keys %{$heap->{groups}->{$g}}) {
            my $event = $heap->{groups}->{$g}->{$r};
            $event .= ($type & REGULAR_MESS) ? "_regular" : "_admin";
            $kernel->post( $r => $event => [$sender, $message, $type, $groups] );
        }
    }
}

sub publish {
    my $groups = $_[ARG0];
    my $message = $_[ARG1];
    my $heap = $_[HEAP];
    my $sender = $_[SENDER];
    my $r_heap = $sender->get_heap();

    Spread::multicast($r_heap->{spread}, RELIABLE_MESS, $groups, 0, $message);
}

sub subscribe {
    my $groups = $_[ARG0];
    my $heap = $_[HEAP];
    my $session = $_[SESSION];
    my $sender = $_[SENDER];
    my $event = $_[ARG1] || $groups;
    my $r_heap = $sender->get_heap();

    $heap->{groups}->{$groups}->{$sender} = $event;

    Spread::join($r_heap->{spread}, $groups);
}

$VERSION;

__END__