Coro/Coro/Socket.pm

=head1 NAME

Coro::Socket - non-blocking socket-io

=head1 SYNOPSIS

 use Coro::Socket;

 # listen on an ipv4 socket
 my $socket = new Coro::Socket PeerHost => "localhost",
                               PeerPort => 'finger';

 # listen on any other type of socket
 my $socket = Coro::Socket->new_from_fh
                 (IO::Socket::UNIX->new
                     Local  => "/tmp/socket",
                     Type   => SOCK_STREAM,
                 );

=head1 DESCRIPTION

This module is an L<AnyEvent> user, you need to make sure that you use and
run a supported event loop.

This module implements socket-handles in a coroutine-compatible way,
that is, other coroutines can run while reads or writes block on the
handle. See L<Coro::Handle>, especially the note about prefering method
calls.

=over 4

=cut

package Coro::Socket;

no warnings "uninitialized";

use strict;

use Errno ();
use Carp qw(croak);
use Socket;
use IO::Socket::INET ();

use Coro::Util ();

use base qw(Coro::Handle IO::Socket::INET);

our $VERSION = 4.8;

our (%_proto, %_port);

sub _proto($) {
   $_proto{$_[0]} ||= do {
      ((getprotobyname $_[0])[2] || (getprotobynumber $_[0])[2])
         or croak "unsupported protocol: $_[0]";
   };
}

sub _port($$) {
   $_port{$_[0],$_[1]} ||= do {
      return $_[0] if $_[0] =~ /^\d+$/;

      $_[0] =~ /([^(]+)\s*(?:\((\d+)\))?/x
         or croak "unparsable port number: $_[0]";
      ((getservbyname $1, $_[1])[2]
        || (getservbyport $1, $_[1])[2]
        || $2)
         or croak "unknown port: $_[0]";
   };
}

sub _sa($$$) {
   my ($host, $port, $proto) = @_;

   $port or $host =~ s/:([^:]+)$// and $port = $1;

   my $_proto = _proto($proto);
   my $_port = _port($port, $proto);

   my $_host = Coro::Util::inet_aton $host
      or croak "$host: unable to resolve";

   pack_sockaddr_in $_port, $_host
}

=item $fh = new Coro::Socket param => value, ...

Create a new non-blocking tcp handle and connect to the given host
and port. The parameter names and values are mostly the same as for
IO::Socket::INET (as ugly as I think they are).

The parameters officially supported currently are: C<ReuseAddr>,
C<LocalPort>, C<LocalHost>, C<PeerPort>, C<PeerHost>, C<Listen>, C<Timeout>.

   $fh = new Coro::Socket PeerHost => "localhost", PeerPort => 'finger';

=cut

sub _prepare_socket {
   my ($self, $arg) = @_;

   $self
}
   
sub new {
   my ($class, %arg) = @_;

   $arg{Proto}     ||= 'tcp';
   $arg{LocalHost} ||= delete $arg{LocalAddr};
   $arg{PeerHost}  ||= delete $arg{PeerAddr};
   defined ($arg{Type}) or $arg{Type} = $arg{Proto} eq "tcp" ? SOCK_STREAM : SOCK_DGRAM;

   socket my $fh, PF_INET, $arg{Type}, _proto ($arg{Proto})
      or return;

   my $self = bless Coro::Handle->new_from_fh (
      $fh,
      timeout       => $arg{Timeout},
      forward_class => $arg{forward_class},
      partial       => $arg{partial},
   ), $class
      or return;

   $self->configure (\%arg)
}

sub configure {
   my ($self, $arg) = @_;

   if ($arg->{ReuseAddr}) {
      $self->setsockopt (SOL_SOCKET, SO_REUSEADDR, 1)
         or croak "setsockopt(SO_REUSEADDR): $!";
   }

   if ($arg->{ReusePort}) {
      $self->setsockopt (SOL_SOCKET, SO_REUSEPORT, 1)
         or croak "setsockopt(SO_REUSEPORT): $!";
   }

   if ($arg->{Broadcast}) {
      $self->setsockopt (SOL_SOCKET, SO_BROADCAST, 1)
         or croak "setsockopt(SO_BROADCAST): $!";
   }

   if ($arg->{LocalPort} || $arg->{LocalHost}) {
      my @sa = _sa($arg->{LocalHost} || "0.0.0.0", $arg->{LocalPort} || 0, $arg->{Proto});
      $self->bind ($sa[0])
         or croak "bind($arg->{LocalHost}:$arg->{LocalPort}): $!";
   }

   if ($arg->{PeerHost}) {
      my @sa = _sa ($arg->{PeerHost}, $arg->{PeerPort}, $arg->{Proto});

      for (@sa) {
         $! = 0;

         if ($self->connect ($_)) {
            next unless writable $self;
            $! = unpack "i", $self->getsockopt (SOL_SOCKET, SO_ERROR);
         }

         $! or last;

         $!{ECONNREFUSED} or $!{ENETUNREACH} or $!{ETIMEDOUT} or $!{EHOSTUNREACH}
            or return;
      }
   } elsif (exists $arg->{Listen}) {
      $self->listen ($arg->{Listen})
         or return;
   }

   $self
}

1;

=back

=head1 AUTHOR

 Marc Lehmann <schmorp@schmorp.de>
 http://home.schmorp.de/

=cut

Revision:	1.34
Committed:	Tue Sep 30 17:12:35 2008 UTC (15 years, 8 months ago) by root
Branch:	MAIN
CVS Tags:	rel-4_8, rel-4_801
Changes since 1.33:	+1 -1 lines
Log Message:	* empty log message *
#	User	Rev	Content
1	root	1.1	=head1 NAME
2
3			Coro::Socket - non-blocking socket-io
4
5			=head1 SYNOPSIS
6
7			use Coro::Socket;
8
9	root	1.14	# listen on an ipv4 socket
10			my $socket = new Coro::Socket PeerHost => "localhost",
11			PeerPort => 'finger';
12
13			# listen on any other type of socket
14			my $socket = Coro::Socket->new_from_fh
15			(IO::Socket::UNIX->new
16			Local => "/tmp/socket",
17			Type => SOCK_STREAM,
18			);
19
20	root	1.1	=head1 DESCRIPTION
21
22	root	1.20	This module is an L<AnyEvent> user, you need to make sure that you use and
23			run a supported event loop.
24
25	root	1.1	This module implements socket-handles in a coroutine-compatible way,
26			that is, other coroutines can run while reads or writes block on the
27	root	1.17	handle. See L<Coro::Handle>, especially the note about prefering method
28			calls.
29	root	1.1
30			=over 4
31
32			=cut
33
34			package Coro::Socket;
35
36	root	1.14	no warnings "uninitialized";
37
38			use strict;
39
40	root	1.1	use Errno ();
41			use Carp qw(croak);
42			use Socket;
43	root	1.14	use IO::Socket::INET ();
44	root	1.1
45	root	1.6	use Coro::Util ();
46
47	root	1.14	use base qw(Coro::Handle IO::Socket::INET);
48
49	root	1.34	our $VERSION = 4.8;
50	root	1.1
51	root	1.14	our (%_proto, %_port);
52	root	1.1
53			sub _proto($) {
54			$_proto{$_[0]} \|\|= do {
55			((getprotobyname $_[0])[2] \|\| (getprotobynumber $_[0])[2])
56			or croak "unsupported protocol: $_[0]";
57			};
58			}
59
60			sub _port($$) {
61	root	1.9	$_port{$_[0],$_[1]} \|\|= do {
62			return $_[0] if $_[0] =~ /^\d+$/;
63
64			$_[0] =~ /([^(]+)\s*(?:\((\d+)\))?/x
65			or croak "unparsable port number: $_[0]";
66			((getservbyname $1, $_[1])[2]
67			\|\| (getservbyport $1, $_[1])[2]
68			\|\| $2)
69	root	1.1	or croak "unknown port: $_[0]";
70			};
71			}
72
73			sub _sa($$$) {
74			my ($host, $port, $proto) = @_;
75	root	1.22
76	root	1.14	$port or $host =~ s/:([^:]+)$// and $port = $1;
77	root	1.22
78	root	1.1	my $_proto = _proto($proto);
79			my $_port = _port($port, $proto);
80
81	root	1.22	my $_host = Coro::Util::inet_aton $host
82			or croak "$host: unable to resolve";
83
84			pack_sockaddr_in $_port, $_host
85	root	1.1	}
86
87	root	1.14	=item $fh = new Coro::Socket param => value, ...
88	root	1.1
89			Create a new non-blocking tcp handle and connect to the given host
90	root	1.21	and port. The parameter names and values are mostly the same as for
91	root	1.1	IO::Socket::INET (as ugly as I think they are).
92
93	root	1.21	The parameters officially supported currently are: C<ReuseAddr>,
94			C<LocalPort>, C<LocalHost>, C<PeerPort>, C<PeerHost>, C<Listen>, C<Timeout>.
95	root	1.1
96	root	1.14	$fh = new Coro::Socket PeerHost => "localhost", PeerPort => 'finger';
97	root	1.1
98			=cut
99
100			sub _prepare_socket {
101	root	1.14	my ($self, $arg) = @_;
102
103			$self
104			}
105
106			sub new {
107			my ($class, %arg) = @_;
108	root	1.1
109	root	1.14	$arg{Proto} \|\|= 'tcp';
110			$arg{LocalHost} \|\|= delete $arg{LocalAddr};
111			$arg{PeerHost} \|\|= delete $arg{PeerAddr};
112			defined ($arg{Type}) or $arg{Type} = $arg{Proto} eq "tcp" ? SOCK_STREAM : SOCK_DGRAM;
113
114			socket my $fh, PF_INET, $arg{Type}, _proto ($arg{Proto})
115	root	1.1	or return;
116
117	root	1.14	my $self = bless Coro::Handle->new_from_fh (
118			$fh,
119			timeout => $arg{Timeout},
120			forward_class => $arg{forward_class},
121			partial => $arg{partial},
122			), $class
123	root	1.1	or return;
124
125	root	1.14	$self->configure (\%arg)
126			}
127
128			sub configure {
129			my ($self, $arg) = @_;
130
131	root	1.1	if ($arg->{ReuseAddr}) {
132	root	1.14	$self->setsockopt (SOL_SOCKET, SO_REUSEADDR, 1)
133	root	1.1	or croak "setsockopt(SO_REUSEADDR): $!";
134			}
135
136			if ($arg->{ReusePort}) {
137	root	1.14	$self->setsockopt (SOL_SOCKET, SO_REUSEPORT, 1)
138	root	1.1	or croak "setsockopt(SO_REUSEPORT): $!";
139			}
140
141	root	1.18	if ($arg->{Broadcast}) {
142			$self->setsockopt (SOL_SOCKET, SO_BROADCAST, 1)
143			or croak "setsockopt(SO_BROADCAST): $!";
144			}
145
146	root	1.14	if ($arg->{LocalPort} \|\| $arg->{LocalHost}) {
147			my @sa = _sa($arg->{LocalHost} \|\| "0.0.0.0", $arg->{LocalPort} \|\| 0, $arg->{Proto});
148			$self->bind ($sa[0])
149	root	1.1	or croak "bind($arg->{LocalHost}:$arg->{LocalPort}): $!";
150			}
151
152	root	1.14	if ($arg->{PeerHost}) {
153			my @sa = _sa ($arg->{PeerHost}, $arg->{PeerPort}, $arg->{Proto});
154	root	1.1
155			for (@sa) {
156			$! = 0;
157
158	root	1.14	if ($self->connect ($_)) {
159			next unless writable $self;
160			$! = unpack "i", $self->getsockopt (SOL_SOCKET, SO_ERROR);
161	root	1.1	}
162
163			$! or last;
164
165			$!{ECONNREFUSED} or $!{ENETUNREACH} or $!{ETIMEDOUT} or $!{EHOSTUNREACH}
166			or return;
167			}
168	root	1.14	} elsif (exists $arg->{Listen}) {
169			$self->listen ($arg->{Listen})
170	root	1.1	or return;
171			}
172
173	root	1.14	$self
174	root	1.1	}
175
176			1;
177
178	root	1.14	=back
179
180	root	1.1	=head1 AUTHOR
181
182	root	1.14	Marc Lehmann <schmorp@schmorp.de>
183			http://home.schmorp.de/
184	root	1.1
185			=cut
186