Creazione e instradamento dinamico della coda di Celery

Question

Sto provando a chiamare un'attività e creare una coda per quell'attività se non esiste, quindi inserire immediatamente in tale coda l'attività chiamata. Ho il seguente codice:

@task 
def greet(name): 
    return "Hello %s!" % name 


def run(): 
    result = greet.delay(args=['marc'], queue='greet.1', 
     routing_key='greet.1') 
    print result.ready() 

poi ho un router personalizzato:

class MyRouter(object): 

    def route_for_task(self, task, args=None, kwargs=None): 
     if task == 'tasks.greet': 
      return {'queue': kwargs['queue'], 
        'exchange': 'greet', 
        'exchange_type': 'direct', 
        'routing_key': kwargs['routing_key']} 
     return None 

questo crea uno scambio chiamato greet.1 e una coda di chiamata greet.1 ma la coda è vuota. Lo scambio dovrebbe essere chiamato semplicemente greet che sa come instradare una chiave di routing come greet.1 alla coda chiamata greet.1.

Qualche idea?

Answer 1

Quando si esegue il seguente:

task.apply_async(queue='foo', routing_key='foobar') 

Poi sedano avrà valori di default dalla coda 'pippo' in CELERY_QUEUES, o se non esiste quindi creare automaticamente utilizzando (coda = foo, scambio = foo, routing_key = foo)

quindi, se 'foo' non esiste in CELERY_QUEUES si finirà con:

queues['foo'] = Queue('foo', exchange=Exchange('foo'), routing_key='foo') 

Il produttore sarà quindi dichiarare che coda, ma dal momento che si ignora il routing_key, in realtà inviare il messaggio utilizzando routing_key = 'foobar'

Questo può sembrare strano, ma il comportamento è effettivamente utile per gli scambi argomento, cui pubblicate a diversi argomenti.

È più difficile fare ciò che si desidera, tuttavia è possibile creare la coda da soli e dichiararla, ma ciò non funzionerà correttamente con i tentativi di pubblicazione automatica dei messaggi. Sarebbe meglio se l'argomento della coda per apply_async potesse supportare una personalizzata kombu.Queue invece che sia dichiarata e utilizzata come destinazione. Forse potresti aprire un problema per questo a http://github.com/celery/celery/issues