Clever Geek Handbook

Multiple watchers working together

I have an input that does a real-time search by user type. For example, suppose he was looking for the following:

car

Result:

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

I was able to do this successfully, here's how:

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);
  }

}

And the view:

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

I want the user to be able to enter multiple words at the same time and perform real-time search for each word separately. For example, suppose he was looking for the following:

car food sun

The result for the car will be:

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

The result for food will be:

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

The result for sun is:

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

And also combine the results of each word, in this case it will look like this:

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

But let's say that a user, after typing in all the words and performing a search, wants to change one of them. Only the search for the word that was changed needs to be redone and the merge of the final result also needs to be redone.

I still don't know all the features of rxjs and I don't know what the ideal way to achieve this would be. If you need a link, Display Purposes has a very similar search engine.

+3


source to share


2 answers


I think you want something like this:



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);
 }
0


source


I came up with this huge but partial solution:

The spell is here: 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));


What to improve:

  • I had to use a custom implementation combineAll

    that didn't wait for the original thread to complete. This can leak memory if not implemented correctly, and it must remove internal subscriptions that are terminating (it is not).
  • Improvement in word recognition needs to be improved: removing a word in the middle makes half of the words to be searched again.

This is perhaps a better, more straight forward solution, but I just could come up with this mess. Feels very hacky / dangerous.

0


source






More articles:


All Articles