lib/MojoX/Gearman.pm

   1 package MojoX::Gearman;
   2
   3 use strict;
   4 use warnings;
   5
   6 our $VERSION = 0.1;
   7 use base 'Mojo::Base';
   8
   9 use Mojo::IOLoop;
  10 use List::Util    ();
  11 use Mojo::Util    ();
  12 use Scalar::Util        ();
  13 use Data::Dump qw(dump);
  14 require Carp;
  15
  16 __PACKAGE__->attr(server   => '127.0.0.1:4730');
  17 __PACKAGE__->attr(ioloop   => sub { Mojo::IOLoop->singleton });
  18 __PACKAGE__->attr(error => undef);
  19 __PACKAGE__->attr(timeout  => 300);
  20 __PACKAGE__->attr(encoding => 'UTF-8');
  21 __PACKAGE__->attr(
  22         on_error => sub {
  23                 sub {
  24                         my $gearman = shift;
  25                         warn "Gearman error: ", $gearman->error, "\n";
  26                   }
  27         }
  28 );
  29
  30 sub DESTROY {
  31         my $self = shift;
  32
  33         # Loop
  34         return unless my $loop = $self->ioloop;
  35
  36         # Cleanup connection
  37         $loop->drop($self->{_connection})
  38           if $self->{_connection};
  39 }
  40
  41 sub connect {
  42         my $self = shift;
  43
  44         # drop old connection
  45         if ($self->connected) {
  46                 $self->ioloop->drop($self->{_connection});
  47         }
  48
  49         $self->server =~ m{^([^:]+)(:(\d+))?};
  50         my $address = $1;
  51         my $port = $3 || 4730;
  52
  53         Scalar::Util::weaken $self;
  54
  55         # connect
  56         $self->{_connecting} = 1;
  57         $self->{_connection} = $self->ioloop->connect(
  58                 {   address     => $address,
  59                         port       => $port,
  60                         on_connect => sub { $self->_on_connect(@_) },
  61                         on_read => sub { $self->_on_read(@_) },
  62                         on_error   => sub { $self->_on_error(@_) },
  63                         on_hup   => sub { $self->_on_hup(@_) },
  64                 }
  65         );
  66
  67         return $self;
  68 }
  69
  70 sub connected {
  71         my $self = shift;
  72
  73         return $self->{_connection};
  74 }
  75
  76 sub req {
  77         my $self = shift;
  78         my $type = shift;
  79         my $data = join("\0", @_);
  80
  81         Mojo::Util::encode($self->encoding, $data) if $self->encoding;
  82
  83         my $ret;
  84
  85         my $cb = sub {
  86                 my ( $self, $data ) = @_;
  87                 $self->ioloop->stop;
  88                 warn "# <<<< ",dump($data);
  89                 my ($magic, $type, $len) = unpack( "a4NN", $data );
  90                 die "wrong magic [$magic]" unless $magic eq "\0RES";
  91                 die "ERROR" if $type == 19;
  92                 $ret = substr($data,12,$len);
  93         };
  94
  95         my $len = length($data);
  96         my $message = pack("a4NN", "\0REQ", $type, length $data ) . $data;
  97         warn "# >>>> ",dump($data);
  98
  99         my $mqueue = $self->{_message_queue} ||= [];
 100         my $cqueue = $self->{_cb_queue}   ||= [];
 101
 102
 103         push @$mqueue, $message;
 104         push @$cqueue, $cb;
 105
 106         $self->connect unless $self->{_connection};
 107         $self->_send_next_message;
 108
 109         $self->ioloop->start;
 110
 111         return $ret;
 112 }
 113
 114 sub start {
 115         my ($self) = @_;
 116
 117         $self->ioloop->start;
 118         return $self;
 119 }
 120
 121 sub stop {
 122         my ($self) = @_;
 123
 124         $self->ioloop->stop;
 125         return $self;
 126 }
 127
 128 sub _send_next_message {
 129         my ($self) = @_;
 130
 131         if ((my $c = $self->{_connection}) && !$self->{_connecting}) {
 132                 while (my $message = shift @{$self->{_message_queue}}) {
 133                 warn "# write ",dump($message);
 134                 $self->ioloop->write($c, $message);
 135                 }
 136         }
 137 }
 138
 139 sub _on_connect {
 140         my ($self, $ioloop, $id) = @_;
 141         delete $self->{_connecting};
 142
 143         $ioloop->connection_timeout($id => $self->timeout);
 144
 145         $self->_send_next_message;
 146 }
 147
 148 sub _on_error {
 149         my ($self, $ioloop, $id, $error) = @_;
 150
 151         $self->error($error);
 152         $self->_inform_queue;
 153
 154         $self->on_error->($self);
 155
 156         $ioloop->drop($id);
 157 }
 158
 159 sub _on_hup {
 160         my ($self, $ioloop, $id) = @_;
 161
 162         $self->{error} ||= 'disconnected';
 163         $self->_inform_queue;
 164
 165         delete $self->{_message_queue};
 166
 167         delete $self->{_connecting};
 168         delete $self->{_connection};
 169 }
 170
 171 sub _inform_queue {
 172         my ($self) = @_;
 173
 174         for my $cb (@{$self->{_cb_queue}}) {
 175                 $cb->($self) if $cb;
 176         }
 177         $self->{_queue} = [];
 178 }
 179
 180 sub _on_read {
 181         my ($self, $ioloop, $id, $data) = @_;
 182
 183         my $cb = shift @{$self->{_cb_queue}};
 184         if ($cb) {
 185                 Mojo::Util::decode($self->encoding, $data) if $data;
 186                 warn "# on read callback with ", dump($data);
 187                 $cb->($self, $data);
 188         } else {
 189                 warn "no callback";
 190         }
 191
 192         # Reset error after callback dispatching
 193         $self->error(undef);
 194 }
 195
 196 1;
 197 __END__
 198
 199 =head1 NAME
 200
 201 MojoX::Gearman - asynchronous Gearman client for L<Mojolicious>.
 202
 203 =head1 SYNOPSIS
 204
 205         use MojoX::Gearman;
 206
 207         my $gearman = MojoX::Gearman->new(server => '127.0.0.1:4730');
 208
 209 =head1 DESCRIPTION
 210
 211 L<MojoX::Gearman> is an asynchronous client to Gearman for Mojo.
 212
 213 =head1 ATTRIBUTES
 214
 215 L<MojoX::Gearman> implements the following attributes.
 216
 217 =head2 C<server>
 218
 219         my $server = $gearman->server;
 220         $gearman         = $gearman->server('127.0.0.1:4730');
 221
 222 C<Gearman> server connection string, defaults to '127.0.0.1:4730'.
 223
 224 =head2 C<ioloop>
 225
 226         my $ioloop = $gearman->ioloop;
 227         $gearman         = $gearman->ioloop(Mojo::IOLoop->new);
 228
 229 Loop object to use for io operations, by default a L<Mojo::IOLoop> singleton
 230 object will be used.
 231
 232 =head2 C<timeout>
 233
 234         my $timeout = $gearman->timeout;
 235         $gearman          = $gearman->timeout(100);
 236
 237 Maximum amount of time in seconds a connection can be inactive before being
 238 dropped, defaults to C<300>.
 239
 240 =head2 C<encoding>
 241
 242         my $encoding = $gearman->encoding;
 243         $gearman           = $gearman->encoding('UTF-8');
 244
 245 Encoding used for stored data, defaults to C<UTF-8>.
 246
 247 =head1 METHODS
 248
 249 =head2 C<req>
 250
 251         $gearman->req( $type, $data );
 252
 253 =head2 C<error>
 254
 255         $gearman->execute("ping" => sub {
 256                 my ($gearman, $result) = @_;
 257                 die $gearman->error unless defined $result;
 258         }
 259
 260 Returns error occured during command execution.
 261 Note that this method returns error code just from current command and
 262 can be used just in callback.
 263
 264 =head2 C<on_error>
 265
 266         $gearman->on_error(sub{
 267                 my $gearman = shift;
 268                 warn 'Gearman error ', $gearman->error, "\n";
 269         });
 270
 271 Executes if error occured. Called before commands callbacks.
 272
 273 =head2 C<start>
 274
 275         $gearman->start;
 276
 277 Starts IOLoop. Shortcut for $gearman->ioloop->start;
 278
 279 =head1 SEE ALSO
 280
 281 L<Gearman::Client>, L<Mojolicious>, L<Mojo::IOLoop>
 282
 283 =head1 SUPPORT
 284
 285 =head1 DEVELOPMENT
 286
 287 =head2 Repository
 288
 289         https://github.com/dpavlin/mojox-gearman
 290
 291 =head1 AUTHOR
 292
 293 Dobrica Pavlinusic, C<dpavlin@rot13.org>.
 294
 295 =head1 COPYRIGHT AND LICENSE
 296
 297 Copyright (C) 2010, Dobrica Pavlinusic
 298
 299 This program is free software, you can gearmantribute it and/or modify it under
 300 the terms of the Artistic License version 2.0.
 301
 302 =cut