одна очередь на паука, correlation_id на запрос...

в сообщениях что будет? урлы для парсинга?

нене, текст выдранный скрапи

скрапи инициатор

еще не хватает знаний об ampq в целом.. хочется отправлять месседж, ждать респонс со своим correlation_id, при перехвате чужого corr_id NACKать его, а при получении своего ответа ACK'ать и возвращать его в пайплайн...

м?м?

работают так с рэббитом? или это борщ?)

Сообщения через ребит, судя по доке получать так

def callback(ch, method, properties, body):
   print(" [x] Received %r" % body)
   time.sleep( body.count('.') )
   print(" [x] Done")
   ch.basic_ack(delivery_tag = method.delivery_tag)

channel.basic_consume(queue='hello', on_message_callback=callback)
При этом после channel.basic_consume(queue='hello', on_message_callback=callback) заводится "демон" который слушает очередь на наличие сообщение и кидает их в callback, у меня с этим проблема была, хотя опыта в питоне и ребит не так много, хз, может сам что то не так делал

time.sleep НЕНЕНЕНЕ

а, это пример

да даже без time.sleep не заведется, так как там твистед реактор будет ругаться на channel.basic_consume

ачо

судя по твистед примеру channel.basic_consume возвращает деферед

судя по доке, впрочем, нет :) возможно это разные

а, наверно это потому что всё основано на pika.adapters.twisted_connection

я вобщем не знаю какая там архитектура предлагается, но если там пабсаб с кроликом, то получается что пайплайн с одной стороны ждёт событий от кролика, а с другой - от паука, и наверно сложить это вместе может быть нетривиально, но решабельно

если process_item должен ждать какого-то события от кролика ("данные пришли" например), то мы делаем деферед, возвращаем его из process_item, и файрим его когда данные придут

придётся навертеть коллбэков, файрящих коллбэки

код из Learning scrapy
class UsingBlocking(object):

 @defer.inlineCallbacks
 def process_item(self, item, spider):
   price = item["price"][0]
   out = defer.Deferred()
   reactor.callInThread(self._do_calculation, price, out)
   item["price"][0] = yield out
   defer.returnValue(item)

 def _do_calculation(self, price, out):
   new_price = price + 1
   time.sleep(0.10)
   reactor.callFromThread(out.callback, new_price)

ну это базовое совсем