Worker job for activity feed

worker.py

#!/usr/bin/env python2.7

import string
import random

import memcache

FRAGMENTATION_THRESHOLD = 3


def encode(keys, operation='+'):
    return ''.join(operation + k + ' ' for k in keys)


def modify(mc, feedName, operation, keys):
    encoded = encode(keys, operation)
    if not mc.append(feedName, encoded):
        if operation == '+':
            mc.add(feedName, encoded)


def add(mc, feedName, *keys):
    modify(mc, feedName, '+', keys)


def remove(mc, feedName, *keys):
    modify(mc, feedName, '-', keys)


def decode(data):
    keys = set()
    fragmentation = 0
    for k in data.split():
        operation, key = k[0], k[1:]
        if operation == '+':
            keys.add(key)
        elif operation == '-':
            keys.discard(key)
            fragmentation += 1

    return fragmentation, keys


def items(mc, feedName, forceCompaction=False):
    data = mc.get(feedName)
    fragmentation, keys = decode(data)

    if forceCompaction or fragmentation > FRAGMENTATION_THRESHOLD:
        compacted = encode(keys)
        # Blocks updates to same blob by multiple processes
        mc.cas(feedName, compacted)

    return keys


def id_generator(size=5, chars=string.digits):
    return ''.join(random.choice(chars) for _ in range(size))


def main():
    mc = memcache.Client(['127.0.0.1:11211'], debug=0)

    key = 'username_' + id_generator()

    for i in xrange(5):
        add(mc, key, 'a', 'b', 'c')
        remove(mc, key, 'c')

    print 'Before compaction key:', key, 'value:', mc.get(key)
    items(mc, key)
    print 'After compaction key:', key, 'value:', mc.get(key)

    mc.delete(key)

if __name__ == '__main__':
    main()