Observable.next() 中断循环

Question

这段代码解析三元组（不是那么重要）并且应该 return 一次一个停车位。问题是调用 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);
    }

  })
})

Answer 1

一般来说，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'); 
}

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.4.2/Rx.js"></script>

每次都会发出 I still get hit through。很可能 observer 在您的代码中未定义，因此调用 observer.next() 会中断并且从未见过以下 console.log。

Answer 2

鉴于您更新的代码，我建议对其进行调试。该错误很可能不在 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;
}

这将使调试正在发生的事情变得更加容易。

Observable.next() 中断循环

Observable.next() breaks loop

rxjs

typescript

rxjs5

angular