bdb-queue.py

import os
import shutil
from bsddb.db import *

DB_HOME = 'data'
DB_FILENAME = 'queue.db'

if os.path.isdir(DB_HOME):
    shutil.rmtree(DB_HOME)
os.makedirs(DB_HOME)

dbenv = DBEnv()
dbenv.set_flags(DB_AUTO_COMMIT, 1)
dbenv.open(DB_HOME, DB_CREATE|DB_INIT_MPOOL|DB_INIT_LOCK|DB_INIT_TXN|DB_REGISTER|DB_RECOVER)

def open_db():
    db = DB(dbenv)
    db.set_flags(DB_INORDER)
    db.set_re_len(768)
    db.set_pagesize(8192)
    db.set_q_extentsize(16 * 1024 * 1024 // 8192)
    db.open(DB_FILENAME, None, DB_QUEUE, DB_CREATE)
    return db

def do_generate(db, count=256):
    data = 'x' * 768
    for i in xrange(count):
        txn = dbenv.txn_begin()
        recid = db.append(data, txn)
        if i % 4 == 3:
            txn.abort()
        else:
            txn.commit()
        print 'Generated 0x%x' % (recid,)

def do_consume(db):
    print 'Berkeley DB v%d.%d.%d' % version()
    while True:
        chunk = db.consume()
        if chunk is None:
            break
        print 'Consumed %r' % (chunk[0],)

db = open_db()
do_generate(db)
do_consume(db)
db.close()