view release on metacpan or  search on metacpan

lib/Danga/Socket/AnyEvent.pm  view on Meta::CPAN

    my ($class, $ref) = @_;

    if (ref $class) {
        # per-object callback
        my Danga::Socket $self = $class;
        if (defined $ref && ref $ref eq 'CODE') {
            $PLCMap{$self->{fd}} = $ref;
        } else {
            delete $PLCMap{$self->{fd}};
        }
    } else {
        # global callback
        $PostLoopCallback = (defined $ref && ref $ref eq 'CODE') ? $ref : undef;
    }
}

# Internal function: run the post-event callback, send read events
# for pushed-back data, and close pending connections.  returns 1
# if event loop should continue, or 0 to shut it all down.
sub PostEventLoop {
    # fire read events for objects with pushed-back read data
    my $loop = 1;
    while ($loop) {
        $loop = 0;
        foreach my $fd (keys %PushBackSet) {
            my Danga::Socket $pob = $PushBackSet{$fd};

            # a previous event_read invocation could've closed a
            # connection that we already evaluated in "keys
            # %PushBackSet", so skip ones that seem to have
            # disappeared.  this is expected.
            next unless $pob;

            die "ASSERT: the $pob socket has no read_push_back" unless @{$pob->{read_push_back}};
            next unless (! $pob->{closed} &&
                         $pob->{event_watch} & POLLIN);
            $loop = 1;
            $pob->event_read;
        }
    }

    # now we can close sockets that wanted to close during our event processing.
    # (we didn't want to close them during the loop, as we didn't want fd numbers
    #  being reused and confused during the event loop)
    while (my $sock = shift @ToClose) {
        my $fd = fileno($sock);

        # close the socket.  (not a Danga::Socket close)
        $sock->close;

        # and now we can finally remove the fd from the map.  see
        # comment above in _cleanup.
        delete $DescriptorMap{$fd};
    }


    # by default we keep running, unless a postloop callback (either per-object
    # or global) cancels it
    my $keep_running = 1;

    # per-object post-loop-callbacks
    for my $plc (values %PLCMap) {
        $keep_running &&= $plc->(\%DescriptorMap, \%OtherFds);
    }

    # now we're at the very end, call callback if defined
    if (defined $PostLoopCallback) {
        $keep_running &&= $PostLoopCallback->(\%DescriptorMap, \%OtherFds);
    }

    return $keep_running;
}

# Internal method to decorate a watcher callback with extra code to install
# the IdleWatcher necessary to run PostEventLoop.
sub _wrap_watcher_cb {
    my ($cb) = @_;

    return sub {
        my $ret = $cb->(@_);
        $IdleWatcher = AnyEvent->idle(
            cb => sub {
                my $keep_running = PostEventLoop();
                $IdleWatcher = undef; # Free this watcher
                $MainLoopCondVar->send unless $keep_running;
            },
        );
        return $ret;
    };
}

#####################################################################
### Danga::Socket-the-object code
#####################################################################

=head2 OBJECT METHODS

=head2 C<< CLASS->new( $socket ) >>

Create a new Danga::Socket subclass object for the given I<socket> which will
react to events on it during the C<EventLoop>.

This is normally (always?) called from your subclass via:

  $class->SUPER::new($socket);

=cut
sub new {
    my Danga::Socket $self = shift;
    $self = fields::new($self) unless ref $self;

    my $sock = shift;

    $self->{sock}        = $sock;
    my $fd = fileno($sock);

    Carp::cluck("undef sock and/or fd in Danga::Socket->new.  sock=" . ($sock || "") . ", fd=" . ($fd || ""))
        unless $sock && $fd;

    $self->{fd}          = $fd;
    $self->{write_buf}      = [];