view release on metacpan or  search on metacpan

lib/Arcus/Client.pm  view on Meta::CPAN

{
  use bytes;
  my %escapes = map { chr($_) => sprintf('%%%02X', $_) } (0x00..0x20, 0x7f..0xff);
  sub _sanitize_method {
    my $key = shift;
    return undef unless defined($key);
    $key =~ s/([\x00-\x20\x7f-\xff])/$escapes{$1}/ge;
    if (length($key) > 200) {
      $key = sha1_hex($key);
    }
    return $key;
  }
}

my $ENCODE = sub {
  my ($conf, $value, $flags) = @_;
  # serialization
  if (ref $value) {
    $value = eval { $conf->{serialize_methods}->[0]->($value) };
    if ($@) {
      warn "failed to serialize.";
    } else {
      $flags |= F_STORABLE;
    }
  }
  # compression
  if (defined($value) and
      defined($conf->{compress_methods}) and
      $conf->{compress_threshold} > 0 and
      $conf->{compress_threshold} <= length($value)) {
    my $c_val = eval { $conf->{compress_methods}->[0]->($value) };
    if ($@) {
      warn "failed to compress.";
    } elsif (length($c_val) < length($value) * $conf->{compress_ratio}) {
      $value = $c_val;
      $flags |= F_COMPRESS;
    }
  }
  return ($value, $flags);
};

my $DECODE = sub {
  my ($conf, $value, $flags) = @_;
  if (defined($flags)) {
    if (defined($value) && ($flags & F_COMPRESS)) {
      $value = eval { $conf->{compress_methods}->[1]->($value) };
      warn "failed to decompress." if ($@);
    }
    if (defined($value) && ($flags & F_STORABLE)) {
      $value = eval { $conf->{serialize_methods}->[1]->($value) };
      warn "failed to deserialize." if ($@);
    }
  }
  return ($value);
};

sub new {
  my ($class, $args) = @_;
  $args->{connect_timeout} //= 1.0; # second
  $args->{io_timeout}      //= 0.8; # second
  $args->{max_thread} //= 64;
  $args->{nowait} = 0; # TODO: Fix a C Client's noreply flags (issue: #3)
  $args->{hash_namespace} = 1;
  $args->{namespace} //= "";
  $args->{serialize_methods} //= [ \&Storable::nfreeze, \&Storable::thaw ];
  $args->{compress_threshold} //= -1;
  $args->{compress_ratio} //= 0.8;
  $args->{compress_methods} //= [ \&Compress::Zlib::memGzip,
                                  \&Compress::Zlib::memGunzip ] if $HAVE_ZLIB;
  my $arcus = $class->SUPER::new($args);
  bless($arcus, $class);
  $arcus_info{$$arcus} = $args;
  POSIX::AtFork->add_to_child(sub {
    $arcus->connect_proxy();
  });
  return $arcus;
}

sub DESTROY {
  my $arcus = shift;
  $arcus->SUPER::DESTROY;
  if ($arcus_info{$$arcus}) {
    delete $arcus_info{$$arcus};
  }
}

sub CLONE {
  my $class = shift;
  foreach my $arcus (keys %arcus_info) {
    $class->SUPER::new($arcus);
  }
}

for my $method ( qw/set add replace/ ) {
  no strict 'refs';
  my $super = 'SUPER::'.$method;
  *{$method} = sub {
    my ($arcus, $key, $value, $exptime) = @_;
    my ($conf, $flags) = ($arcus_info{$$arcus}, 0);
    return undef unless $conf;
    $key = $SANITIZE->($key);
    ($value, $flags) = $ENCODE->($conf, $value, $flags);
    return $arcus->$super($key, $value, $exptime, $flags);
  };
}

sub cas {
  my ($arcus, $key, $cas, $value, $exptime) = @_;
  my ($conf, $flags) = ($arcus_info{$$arcus}, 0);
  return undef unless $conf;
  $key = $SANITIZE->($key);
  ($value, $flags) = $ENCODE->($conf, $value, $flags);
  return $arcus->SUPER::cas($key, $cas, $value, $exptime, $flags);
}

sub cas_multi {
  my ($arcus, @kvs) = @_;
  my $ctx = wantarray;
  my $conf = $arcus_info{$$arcus};
  return undef unless $conf;
  my (@skvs, @ref);