angular - multiple - Múltiples observables trabajando juntos

rxjs subscribe to multiple observables

Question

Tengo una entrada que, a medida que el usuario escribe, realiza una búsqueda en tiempo real. Por ejemplo, supongamos que ha buscado lo siguiente:

coche

El resultado sería:

[
  {
    id: "1"
    name: "Ferrari"
  },
  {
    id: "2"
    name: "Porsche"
  }
]

Esto lo pude hacer con éxito, así es cómo:

class WordComponent {
  word: Subject<string> = new Subject<string>();
  result: any[] = [];

  constructor(private http: Http) {
    this.subscribe();
  }

  subscribe(): void {
    this.word.debounceTime(400)
      .distinctUntilChanged()
      .switchMap((word: string): Observable<any[]> => this.http.get(word))
      .subscribe((result: any[]): any[] => this.result = result);
  }

  search(event: any): void {
    this.word.next(event.target.value);
  }

}

Y la vista:

<input type="text" placeholder="Word" (keyup)="search($event)">

Quiero que el usuario pueda escribir varias palabras al mismo tiempo y realizar una búsqueda en tiempo real de cada palabra por separado. Por ejemplo, supongamos que ha buscado lo siguiente:

sol de comida del coche

El resultado para el automóvil sería:

[
  {
    id: "1"
    name: "Ferrari"
  },
  {
    id: "2"
    name: "Porsche"
  }
]

El resultado para la comida sería:

[
  {
    id: "3"
    name: "egg"
  },
  {
    id: "4"
    name: "cheese"
  }
]

El resultado para el sol sería:

[
  {
    id: "5"
    name: "star"
  },
  {
    id: "6"
    name: "sky"
  }
]

Y también fusiona los resultados de cada palabra, en este caso se vería así:

[
  [{
      id: "1"
      name: "Ferrari"
    },
    {
      id: "2"
      name: "Porsche"
    }
  ],
  [{
      id: "3"
      name: "egg"
    },
    {
      id: "4"
      name: "cheese"
    }
  ],
  [{
      id: "5"
      name: "star"
    },
    {
      id: "6"
      name: "sky"
    }
  ]
]

Pero digamos que el usuario, después de escribir todas las palabras y realizar la búsqueda, desea cambiar una de ellas. Solo se debe rehacer la búsqueda de la palabra que se modificó, y la fusión del resultado final también deberá rehacerse.

Todavía no conozco todas las características de rxjs y no sé cuál sería la forma ideal de lograrlo. Si desea una referencia, el sitio Display Purposes tiene un motor de búsqueda muy similar.

Answer 1

Creo que necesitas algo como esto:

subscribe(): void {
  this.word.debounceTime(400)
  .distinctUntilChanged()
  .switchMap((words: string): Observable<any[]> => 
    Observable.forkJoin(words.split('' '').map(word => this.http.get(word))) 
  )
  .map(arrayWithArrays => [].concat(arrayWithArrays)
  .subscribe((result: any[]): any[] => this.result = result);
 }

Answer 2

He encontrado esta solución enorme pero parcial:

Violín aquí: https://jsfiddle.net/mtawrhLs/1/

Rx.Observable.prototype.combineAllCont = function () {
    let lastValues = [];

    return Rx.Observable.create(obs => {
        let i = 0;

        let subscription = this.subscribe(stream => {
            const streamIndex = i;
            subscription.add(stream.subscribe(res => {
                lastValues[streamIndex] = res;
                obs.next(lastValues);
            }));
            i++;
        });

        return subscription;
    });
}

/** STUFF TO DEMO **/
let searchBox = [
    '''',
    ''car'',
    ''car '',
    ''car food'',
    ''car food sun'',
    ''cat food sun'',
    ''cat sun''
]

function performSearch(word) {
    return Rx.Observable.defer(() => {
        console.log(''sending search for '' + word);
        return Rx.Observable.of({
            word: word,
            results: word.split('''')
        });
    })
}

let searchBoxStream = Rx.Observable.interval(500)
    .take(searchBox.length * 2)
    .map((i) => searchBox[i % searchBox.length]);

/** GOOD STUFF STARTS HERE **/
let wordsStream = searchBoxStream
    .map((str) => str.trim().split('' '').filter((w) => w.trim() != ''''));

let wordSearchSubjects = [];
let wordSearchStreamSubject = new Rx.ReplaySubject(1);

wordsStream.subscribe((words) => {
    const nWords = words.length;
    const nSubjects = wordSearchSubjects.length;
    // Update streams
    for (i = 0; i < nWords && i < nSubjects; i++) {
        wordSearchSubjects[i].next(words[i]);
    }

    // Create streams
    for (i = nSubjects; i < nWords; i++) {
        const wordSearchSubject = new Rx.ReplaySubject(1);
        wordSearchSubjects.push(wordSearchSubject);
        wordSearchStreamSubject.next(
            wordSearchSubject.asObservable()
                .distinctUntilChanged()
                .flatMap((w) => performSearch(w))
                .concat(Rx.Observable.of(false)) // Ending signal
        )
        wordSearchSubjects[i].next(words[i]);
    }

    // Delete streams
    for (i = nWords; i < nSubjects; i++) {
        wordSearchSubjects[i].complete();
    }
    wordSearchSubjects.length = nWords;
});

let wordSearchStream = wordSearchStreamSubject
    .combineAllCont()
    .map((arr) => arr.filter((r) => r !== false));

resultingStream = wordSearchStream
    .map((arr) => {
        let ret = [];
        arr.forEach(search => {
            ret.push(search.results);
        });
        return ret;
    })
    .subscribe((arr) => console.log(arr));

Cosas para mejorar:

1. Tuve que usar una combineAll personalizada. combineAll implementación que no espera la transmisión original para completarse. Esto puede proporcionar pérdidas de memoria si no se implementa correctamente, y debe eliminar las suscripciones internas que se completan (no lo hace ahora)
2. La detección de cambio de palabras debe mejorarse: si quita una palabra del medio, la mitad de las palabras se buscará nuevamente.

Probablemente hay una solución mejor y más directa, pero podría salir con este lío. Se siente muy hacky / peligroso.