view release on metacpan or  search on metacpan

CAB/Socket.pm  view on Meta::CPAN

  my $flags0 = $s->flags;
  if (!($flags0 & O_NONBLOCK)) {
    fcntl($s->{fh}, F_SETFL, $flags0 | O_NONBLOCK)
      or $s->logconfess("canwrite(): could not set O_NOBLOCK on socket: $!");
  }
  my $wbits = fhbits($s->{fh});
  my $nfound = select(undef, $wbits, undef, $timeout);
  if (!($flags0 & O_NONBLOCK)) {
    fcntl($s->{fh}, F_SETFL, $flags0)
      or $s->logconfess("canwrite(): could not reset socket flags: $!");
  }
  return $nfound;
}

## $s = $s->waitr()
##  + waits indefinitely for input; wrapper for $s->canread(undef)
sub waitr {
  return $_[0]->canread(undef) ? $_[0] : undef;
}

## $bool = $s->waitw()
##  + wrapper for $s->canwrite(undef)
sub waitw {
  return $_[0]->canwrite(undef) ? $_[0] : undef;
}

##==============================================================================
## Server Methods

## $class = $CLASS_OR_OBJECT->clientClass()
##  + default client class, used by newClient()
sub clientClass {
  return ref($_[0]) || $_[0];
}

## $client = $CLASS_OR_OBJECT->newClient(%args)
##  + wrapper for clients, called by $s->accept()
##  + default just calls $CLASS_OR_OBJECT->clientClass->new(%args)
sub newClient {
  my $that = shift;
  return $that->clientClass->new(@_);
}

## $cli_or_undef = $s->accept()
## $cli_or_undef = $s->accept($timeout_secs)
##  + accept incoming client connections with optional timeout
##  + if a client connection is available, it will be returned with $s->newClient(fh=>$fh)
##  + otherwise, if no connection is available, undef will be returned
sub accept {
  my $s = shift;
  my $timeout = @_ ? shift : $s->{timeout};
  if (!defined($timeout) || $s->canread($timeout)) {
    my $cfh = $s->{fh}->accept();
    return undef if (!defined($cfh));
    return $s->newClient(fh=>$cfh);
  }
  return undef;
}

## $rc = $qs->handleClient($cli)
## $rc = $qs->handleClient($cli, %callbacks)
##  + handle a single client request
##  + each client request is a STRING message (command)
##    - request arguments (if required) are sent as separate messages following the command request
##    - server response (if any) depends on command sent
##  + this method parses client request command $cmd and dispatches to
##    - the function $callbacks{lc($cmd)}->($qs,$cli,\$cmd), if defined
##    - the method $qs->can("handle_".lc($cmd))->($qs,$cli,\$cmd), if available
##    - the function $callbacks{DEFAULT}->($qs,$cli,\$cmd), if defined
##    - the method $qs->can("handle_DEFAULT")->($qs,$cli,\$cmd)
##  + returns whatever the handler subroutine does
sub handleClient {
  my ($qs,$cli,%callbacks) = @_;
  my $creq = $cli->get();
  $qs->vlog($qs->{logRequest}, "client request: $$creq");
  if (!ref($creq) || ref($creq) ne 'SCALAR' || ref($$creq)) {
    $qs->logconfess("could not parse client request");
  }
  my $cmd = lc($$creq);
  my ($sub);
  if (defined($sub=$callbacks{$cmd})) {
    return $sub->($qs,$cli,$creq);
  }
  elsif (defined($sub=$qs->can("handle_${cmd}"))) {
    return $sub->($qs,$cli,$creq);
  }
  elsif (defined($sub=$callbacks{DEFAULT})) {
    return $sub->($qs,$cli,$creq);
  }
  elsif (defined($sub=$qs->can("handle_DEFAULT"))) {
    return $sub->($qs,$cli,$creq);
  }
  ##-- should never get here
  $qs->logconfess("could not dispatch client request $$creq");
  return undef;
}

##--------------------------------------------------------------
## Server Methods: Request Handling

## undef = $qs->handle_DEFAULT($cli,\$cmd)
##  + default implementation just logcluck()s and returns undef
sub handle_DEFAULT {
  $_[0]->logcluck("cannot handle client client request ${$_[2]}");
  return undef;
}

##==============================================================================
## Protocol
##  + all socket messages are of the form pack('NN/a*', $flags, $message_data)
##  + $flags is a bitmask of DTA::CAB::Socket flags ($sf_* constants)
##  + length element (second 'N' of pack format) is always 0 for serialized references
##  + $message_data is one of the following:
##    - if    ($flags & $sf_ref)   -> a reference written with nstore_fd(); will be decoded
##    - elsif ($flags & $sf_u8)    -> a UTF-8 encoded string; will be decoded
##    - elsif ($flags & $sf_undef) -> a literal undef value
##    - elsif ($flags & $sf_eoq)   -> undef as end-of-queue marker

##--------------------------------------------------------------
## Protocol: Constants
our $sf_eoq   = 0x1;
our $sf_undef = 0x2;
our $sf_u8    = 0x4;
our $sf_ref   = 0x8;

##--------------------------------------------------------------
## Protocol: Write

## $s = $s->put_header($flags,$len)
##  + write a message header to the socket
sub put_header {
  $_[0]->vtrace("put_header", @_[1..$#_]);
  syswrite($_[0]{fh}, pack('NN', @_[1,2]), 8)==8
    or $_[0]->logconfess("put_header(): could not write message header to socket: $!");
  return $_[0];
}

## $s = $s->put_data(\$data, $len)
## $s = $s->put_data( $data, $len)
##  + write some raw data bytes to the socket (header should already have been sent)
sub put_data {
  $_[0]->vtrace("put_data", @_[1..$#_]);
  return if (!defined($_[0]));
  use bytes;
  my $ref = ref($_[1]) ? $_[1] : \$_[1];
  my $len = defined($_[2]) ? $_[2] : length($$ref);
  if ($len > 0) {