Доступ источник наблюдаемым извне Rx.Observable.prototype

Question

У меня возникают проблемы, выяснить, как получить доступ к источнику наблюдаемому в этой схеме (только пытается выяснить, как это без изменения Rx.Observable.prototype):

 q.drain() 
     .flatMap(function(val){ 
      return q.backpressure(val, function(cb){ 
        setTimeout(cb,1000); 
      }); 
     }) 

Мы называем противодавление в качестве метода на прототипе очереди:

Queue.prototype.backpressure = function(val, fn){ 

    const source = ? // I don't know how to access the source observable... 

    return Rx.Observable.create(sub => { 

     return source.subscribe(val => { 

       fn.call(source, val, function(err, val){ 
        if(err){ 
         sub.error(err); 
        } 
        else{ 
         sub.next(val); 
        } 

       }); 
      }, 
      // be sure to handle errors and completions as appropriate and 
      // send them along 
      err => sub.error(err), 
      () => sub.complete()); 

    }); 
}; 

, но проблема в том, я не знаю, могу ли я получить доступ к источнику наблюдаемую в этой схеме - правильное значение источника, безусловно, не значение this внутри прототипа, поскольку оно принадлежит экземпляру очереди. Моя единственная надежда, я думаю, - это каким-то образом передать источник, наблюдаемый непосредственно в метод противодавления. Кто-нибудь знает, как я могу это сделать? Я не против помещать эту функцию в другое место, она не должна быть методом в очереди, но я думаю, что одна и та же проблема будет существовать в любом случае.

Если это помогает, значению this внутри функции flatMap (если использовать обычную функцию вместо функции стрелки) является объектом MergeMapSubcriber см:

Однако после экспериментов, Я не считаю, что значение MergeMapSubcriber является тем, которое я хочу использовать в качестве источника; мой источник должен быть наблюдаемым TMK, а не подписчиком.

Answer 1

Вы думаете о том, чтобы положить его на прототип Observable?

Observable.prototype.backpressure = function(queue, fn){ 
    const source = this; 

    return this.flatMap(function(val){ 
    return Rx.Observable.create(sub => { 

     return source.subscribe... 
    }); 
    }) 
}; 

Тогда для очереди:

q.drain() 
    .backpressure(q, function(cb) { 
    setTimeout(cb,1000); 
    });