Cache-Memcached-Queue

 view release on metacpan or  search on metacpan

lib/Cache/Memcached/Queue.pm  view on Meta::CPAN

    }

    #checar se é necessário a serialização
    if(ref($value)){
      #serializar
      my $serialized = $self->serializer->serialize($value);
      $value = $serialized;
      undef $serialized;
    }
    $self->load;
    if(!$self->_is_locked || $self->_unlock){
      $self->_lock;
      my $size = $self->size // 0;
      #checando se a fila esta cheia
      if($self->max_enq > 0 && $self->size >= $self->max_enq){
        say "Queue is full!";
      }
      else {
        my $last = $1 if $self->last =~ /_(\d+)$/ // 1;
        #checando se last == first e se existe algum valor
        my $first_value = $self->memcached->get($self->first);

lib/Cache/Memcached/Queue.pm  view on Meta::CPAN

          $last++;
        }
        $size++;
        my $new_last = $self->qid . $last;
        $self->last($new_last);
  
        $self->memcached->set($new_last,$value,0);
      }
      $self->size($size);
      $self->_save(['last','size']);
      $self->_unlock if($self->_is_locked);
    }
  }
  return $ok;
}







sub deq {
  my ( $self, ) = @_;
  my ( $last_item,$value ) = ( undef,undef );
  $self->load;
  if(!$self->_is_locked || $self->_unlock ){
    $self->_lock;
    my $size = $self->size;
    if(!$size){
      say 'Queue is empty!';
    }
    else { 
      my $first = $1 if $self->first =~ /_(\d+)$/ // 1;
      $value = $self->memcached->get($self->first) // '';
      if($value =~ /^\^.*?Storable/i){
        my $unserialized = $self->serializer->deserialize($value);

lib/Cache/Memcached/Queue.pm  view on Meta::CPAN

      }
      else {
        $size = 0;
        $self->first($self->qid . '1',0);
        $self->last($self->qid . '1',0);
        $self->_save(['last']);
      }
    }
    $self->size($size);
    $self->_save(['first','size']);
    $self->_unlock if($self->_is_locked);
  }
  return $value // '';
}






sub show {

lib/Cache/Memcached/Queue.pm  view on Meta::CPAN

sub _lock {
  my ($self,$pid,$lock_pid) = (shift,$$,0);
  $self->load;
  my $qid = $self->qid;
  confess "Panic! No 'qid'!" if (!defined($qid) || !$qid);
  my $lock_idx = $qid . 'LOCKED';
  $lock_pid = $self->_is_locked($lock_idx);
  if(!$lock_pid){
    my $rs = $self->memcached->set($lock_idx,$pid,0);
    confess "Memcached server can't write!" if !defined($rs);
    $lock_pid = $pid;
  }
  else {
    say "is already locked!";
    $lock_pid = 0;
  }
  $self->load;
  return $lock_pid || 0;
}






sub _unlock {
  my ($self,$pid,$ok) = (shift,$$,0);
  $self->load;
  my $qid = $self->qid;
  confess "Panic! No 'qid'!" if (!defined($qid) || !$qid);
  my $lock_idx = $qid . 'LOCKED';
  my $lock_pid = $self->_is_locked($lock_idx);
  if($lock_pid && $lock_pid == $pid){
    my $rs = $self->memcached->set($lock_idx,0,0);
    confess "Memcached can't write!" if !defined($rs);
    $ok = 1;
  }
  elsif($lock_pid && $lock_pid != $pid){
    say "Is locked by another process! $lock_pid";

  }
  $self->load;
  return $ok;
}






sub _is_locked {
  my ($self,$lock_idx) = @_;
  $lock_idx = 0 if !defined $lock_idx;
  my $found = 0;

#  confess "Parameter 'lock_idx' is mandatory!" if (!defined($lock_idx) || !$lock_idx);
  if(!defined($lock_idx) || !$lock_idx){
    $lock_idx = $self->qid . 'LOCKED';
  }
  my $lock_pid = $self->memcached->get($lock_idx); #this pid locked the queue!
#  $lock_pid = 0 if $$ == $lock_pid;
#  foreach my $p(@{$t->table}){
#    if($p->pid == $lock_pid){
#      $found = $p->pid;
#      last;
#    }
#  }
#  $lock_pid = 0 if !$found;
  return $lock_pid ;
}