handlers.py

# coding: utf-8
from __future__ import print_function

# kwargs にドメイン名による名前空間作っていろいろさせると捗る


def printer(x, *args, **kwargs):
    print(x)


def printer_square(x, *args, **kwargs):
    print(x * x)

iwrap.py

# coding: utf-8
from __future__ import unicode_literals
from celery import Celery
import handlers

worker = Celery('tasks', broker='amqp://guest@localhost//')


def single(x):
    trace.delay(x)
    return x


def iwrap(L):
    for x in L:
        single(x)
        yield x


@worker.task
def trace(something, *args, **kwargs):
    attrs = ((n, getattr(handlers, n)) for n in dir(handlers))
    callables = (p for n, p in attrs if not n.startswith('_') and callable(p))
    for fn in callables:
        fn(something, *args, **kwargs)

sample.py

# coding: utf-8
from iwrap import iwrap

for x in iwrap(xrange(1, 10)):
    pass

"例えば" Twitter のユーザストリームからデータを取得してなにかしたい時、
データの入口は 1 つ、出口（使途）は複数ということが多い。

stream の方は取得に専念して、解釈については 各々の handler に任せるべきだ。

そんな時はこのパターンが使える。

csv ファイルをイテレーションするとき（未知のテキストファイルでも良い）
"line か rows を入れると列の傾向を解析する" handler を作ると便利かもしれない。

もちろん、 kwargs などでどの位置から挿入されたのかは知りたい。

もっと端的にいうと、ループはループに専念したい、というケースで有利だ。
行の扱いを他人任せにできる。
ループの結果や順序に対して何かをするケースには全く向いていない。

tee

http://celery.readthedocs.org/en/latest/getting-started/brokers/sqs.html
Celery には Amazon SQS が使えるらしい…。

こりゃいいね。