비동기 구독자 기능으로 관찰 할 수있는 RxJS

Question

나는 직설적 인 것처럼 느껴지지만 놀랍게도 어려운 일을하려고합니다.

나는 RabbitMQ 대기열에 가입하는 기능이 있습니다. 구체적으로 여기 Channel.consume 함수입니다 : http://www.squaremobius.net/amqp.node/channel_api.html#channel_consume

서브 스크립 션 ID로 해결되는 약속 - 나중에 구독을 취소하는 데 필요함 - 또한 메시지가 대기열에서 빠져 나올 때 호출 할 콜백 인수가 있음 .

대기열에서 수신 거부하고 싶을 때 여기 Channel.cancel 함수를 사용하여 소비자를 취소해야합니다 (http://www.squaremobius.net/amqp.node/channel_api.html#channel_cancel). 이것은 이전에 리턴 된 subscription ID를 사용합니다.

관찰 가능 항목을 구독 할 때 큐에 구독하는 Observable에서이 모든 내용을 감싸고 싶습니다. 관찰 가능 항목의 구독이 취소되었을 때 구독을 취소합니다. 그러나 이는 호출의 '이중 비동기'특성으로 인해 다소 어려움이 있습니다 (콜백과 약속 모두 반환한다는 의미입니다).

적으로는, 내가 쓸 수 있도록하고 싶습니다 코드입니다 :이 생성자는 비동기 가입자의 기능 또는 해체의 논리를 지원하지 않는

return new Rx.Observable(async (subscriber) => { 
    var consumeResult = await channel.consume(queueName, (message) => subscriber.next(message)); 
    return async() => { 
    await channel.cancel(consumeResult.consumerTag); 
    }; 
}); 

그러나 이것은 불가능하다.

나는 이것을 이해하지 못했습니다. 내가 여기서 뭔가를 놓치고 있니? 왜 그렇게 어려운거야?

건배, 알렉스

Answer 1

만든 channel.consume 약속 해결하기 위해 단지 함수 내에서 호출되는 관찰자 (이 통과있어 관찰자가 아닌 가입자)로, 기다릴 필요가 없습니다 관찰 당신이 제공합니다.

그러나 사용자가 반환하는 수신 거부 기능은 해당 약정이 해결 될 때까지 기다려야합니다. 그리고 내부적으로 다음과 같이 할 수 있습니다.

return new Rx.Observable((observer) => { 
    var consumeResult = channel.consume(queueName, (message) => observer.next(message)); 
    return() => { 
    consumeResult.then(() => channel.cancel(consumeResult.consumerTag)); 
    }; 
});