Observable.next() 打破循环答案

【问题标题】：Observable.next() breaks loopObservable.next() 打破循环
【发布时间】：2017-07-25 11:07:34
【问题描述】：

这段代码解析三元组（不是那么重要），应该一次返回一个停车位。问题是调用observer.next() 会中断循环，因为它只运行一次。 console.log 也不会被调用。我可能遗漏了一些东西，但是observer.next() 是否有可能打破循环？是有错误还是功能还是我完全错了？

 return Observable.create(observer => {
  this.fetch.get(datasetUrl).then(response => {
    // Get all subjects that are parkings
    const parkingTriples = [],
      parkings = [],
      totalspacesParking = [],
      labels = [];
    for (let index = 0; index < response.triples.length; index++) {
      if (response.triples[index].object === 'http://vocab.datex.org/terms#UrbanParkingSite') {
        parkingTriples.push(response.triples[index]);
      }
      if (response.triples[index].predicate === 'http://vocab.datex.org/terms#parkingNumberOfSpaces') {
        totalspacesParking.push(response.triples[index]);
      }
      if (response.triples[index].predicate === 'http://www.w3.org/2000/01/rdf-schema#label') {
        labels.push(response.triples[index]);
      }
    }
    const _parkings = [];
    for (let index = 0; index < parkingTriples.length; index++) {
      const totalspacesresult = find(totalspacesParking, (o) => {
        return o.subject === parkingTriples[index].subject
      });
      const totalspaces = parseInt(n3.Util.getLiteralValue(totalspacesresult.object), 10);
      const labelresult = find(labels, (o) => {
        return o.subject === parkingTriples[index].subject
      });
      const rdfslabel = n3.Util.getLiteralValue(labelresult.object);
      const id = rdfslabel.replace(' ', '-').toLowerCase();
      observer.next(new Parking(rdfslabel, parkingTriples[index].subject, id, totalspaces, datasetUrl));
      console.log(observer);
    }

  })
})

【问题讨论】：

你能简化你的问题/提供一个测试用例吗？一般来说observer.next() 不应该破坏你的代码
@MarkvanStraten for (let index = 0; index < parkingTriples.length; index++) { observer.next(); console.log('I still get hit through'); }
observer 的签名是什么？当使用常规 Rx.Subject 时，您的简化测试用例将起作用
@MarkvanStraten 我更新了源代码

标签： angular typescript rxjs rxjs5

【解决方案1】：

鉴于您更新的代码，我建议对其进行调试。该错误很可能不在 Rx 位中。我已经重构了您的代码以在 Rx 逻辑和响应解析之间进行拆分：

function getParkings(datasetUrl){
  return Rx.Observable.defer(() => this.fetch.get(datasetUrl))
    .mergeMap(response => parseParkingResponse(response));
}

function parseParkingResponse(response) {
  const parkingTriples = [],
    parkings = [],
    totalspacesParking = [],
    labels = [];
  for (let index = 0; index < response.triples.length; index++) {
    if (response.triples[index].object === 'http://vocab.datex.org/terms#UrbanParkingSite') {
      parkingTriples.push(response.triples[index]);
    }
    if (response.triples[index].predicate === 'http://vocab.datex.org/terms#parkingNumberOfSpaces') {
      totalspacesParking.push(response.triples[index]);
    }
    if (response.triples[index].predicate === 'http://www.w3.org/2000/01/rdf-schema#label') {
      labels.push(response.triples[index]);
    }
  }

  const _parkings = [];
  for (let index = 0; index < parkingTriples.length; index++) {
    const totalspacesresult = find(totalspacesParking, (o) => {
      return o.subject === parkingTriples[index].subject
    });
    const totalspaces = parseInt(n3.Util.getLiteralValue(totalspacesresult.object), 10);
    const labelresult = find(labels, (o) => {
      return o.subject === parkingTriples[index].subject
    });
    const rdfslabel = n3.Util.getLiteralValue(labelresult.object);
    const id = rdfslabel.replace(' ', '-').toLowerCase();
    _parkings.push(new Parking(rdfslabel, parkingTriples[index].subject, id, totalspaces, datasetUrl));
  }

  return _parkings;
}

这将更容易调试正在发生的事情。

【讨论】：

此代码有效！但它正在流式传输阵列。我想一个接一个地流式传输每个停车并在循环本身中调用 oberserver.next(new Parking())
除非您对停车项目进行异步操作，否则parseParkingResponse 将同步，因此流式传输它们不会有任何不同

【解决方案2】：

通常Subject 不会破坏您的代码。鉴于以下情况：

const observer = new Rx.Subject();
const parkingTriples = new Array(10);

observer.subscribe(val => console.log('value emitted: ' + val));

for (let index = 0; index < parkingTriples.length; index++) {
  observer.next(index); 
  console.log('I still get hit through'); 
}

&lt;script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.4.2/Rx.js"&gt;&lt;/script&gt;

这将每次发出I still get hit through。 observer 很可能在您的代码中未定义，因此调用 observer.next() 会中断，并且永远不会看到以下 console.log。

【讨论】：