Directory Based Queue

consumer.pl

use strict;
use warnings;
use Directory::Queue::Simple;

my $dirq = Directory::Queue::Simple->new(path => "/tmp/test");
my $done = 0;
for(my $name = $dirq->first(); $name; $name = $dirq->next()) {
    next unless $dirq->lock($name);
    my $data = $dirq->get($name);
    printf("Body: \"%s\"\n", $data);
    $dirq->remove($name);
    $done += 1;
}
printf("Consumed %s elements\n", $done);

consumer.py

from dirq.QueueSimple import QueueSimple
from random import randint
import time

path = '/tmp/test'

print("start consuming...")
dirq = QueueSimple(path)
done = 0
while True:
    for name in dirq:
        # print("element: %s %s" % (path, name))
        if not dirq.lock(name):
            #print("couldn't lock: %s" % name)
            # name = dirq.next()
            continue
        element = dirq.get(name)
        #print(element.keys())
        print("Body: \"%s\"" % element)
        if randint(1,2) % 2:
            done += 1
            dirq.remove(name)
            #print('Removed')
        else:
            dirq.unlock(name)
        #name = dirq.next()
    print("consumed %i elements" % done)
    total_left = dirq.count()
    print("elements left in the queue: %d" % total_left)
    time.sleep(0.5)

producer.pl

use strict;
use warnings;
use Directory::Queue::Simple;

my $dirq = Directory::Queue::Simple->new(path => "/tmp/test");
foreach my $count (1 .. 100) {
    my $name = $dirq->add("element $count");
    printf("# added element %d as %s\n", $count, $name);
}

producer.py

from dirq.QueueSimple import QueueSimple

# total number of elements
COUNT = 40
# queue head directory
path = 'tmp'

# ========
# PRODUCER
print("*** PRODUCER")
dirq = QueueSimple(path)

done = 1
while done <= COUNT:
    name = dirq.add('The content %i' % done)
    print("added %.2i: %s" % (done, name))
    done += 1

total_stored = dirq.count()
print("total elements in the queue: %d" % total_stored)
del dirq
print('=' * 25)