cont3xt/vueapp/src/components/services/Cont3xtService.js
/**
   * Fetches integration data from the server given a search term.
   * Watches for a stream of data then reads and parses each chunk.
   * Sends chunks back to the subscriber of the search.
   * @param {String} searchTerm - The query to search the integrations for
   * @param {Boolean} skipCache - Whether to use the cached result or fetch it new
   * @returns {Observable} - The observable object to subscribe to updates
   */
Fetches integration data from the server given a search term.
Watches for a stream of data then reads and parses each chunk.
Sends chunks back to the subscriber of the search.
@param {String} searchTerm - The query to search the integrations for
@param {Boolean} skipCache - Whether to use the cached result or fetch it new
@returns {Observable} - The observable object to subscribe to updates
Fetches integration data from the server given a search term . Watches for a stream of data then reads and parses each chunk . Sends chunks back to the subscriber of the search . @param { String } searchTerm - The query to search the integrations for @param { Boolean } skipCache - Whether to use the cached result or fetch it new @returns { Observable } - The observable object to subscribe to updates
search ({ searchTerm, skipCache }) {
    return new Observable((subscriber) => {
      searchTerm = dr.refang(searchTerm.trim());
      if (!searchTerm) { 
        return subscriber.complete();
      }
      const doIntegrations = [];
      for (const integration of store.state.selectedIntegrations) {
        if (store.state.integrations[integration] && store.state.integrations[integration].doable) {
          doIntegrations.push(integration);
        }
      }
      fetch('api/integration/search', {
        method: 'POST',
        headers: setReqHeaders({ 'Content-Type': 'application/json' }),
        body: JSON.stringify({ query: searchTerm, skipCache, doIntegrations })
      }).then((response) => {
        if (!response.ok) { 
          throw new Error(response.statusText);
        }
        return response.body;
      }).then((rStream) => {
        const reader = rStream.getReader();
        const sendChunk = this.sendChunk;
        const decoder = this.decoder;
        return new ReadableStream({
          start () {
            let remaining = '';
            function read () { 
              reader.read().then(({ done, value }) => {
                if (done) { 
                  if (remaining.length) {
                    sendChunk(subscriber, remaining);
                  }
                  return subscriber.complete();
                }
                remaining += decoder(value);
                let pos = 0;
                while ((pos = remaining.indexOf('\n')) > -1) {
                  sendChunk(subscriber, remaining.slice(0, pos - 1));
                  remaining = remaining.slice(pos + 1, remaining.length);
                }
                read(); 
              });
            }
            read();
          }
        });
      }).catch((err) => { 
        subscriber.error(err);
        return subscriber.complete();
      });
    });
  }
search ( { searchTerm , skipCache } ) { return new Observable ( ( subscriber ) => { searchTerm = dr . refang ( searchTerm . trim ( ) ) ; if ( ! searchTerm ) { return subscriber . complete ( ) ; } const doIntegrations = [ ] ; for ( const integration of store . state . selectedIntegrations ) { if ( store . state . integrations [ integration ] && store . state . integrations [ integration ] . doable ) { doIntegrations . push ( integration ) ; } } fetch ( 'api/integration/search' , { method : 'POST' , headers : setReqHeaders ( { 'Content-Type' : 'application/json' } ) , body : JSON . stringify ( { query : searchTerm , skipCache , doIntegrations } ) } ) . then ( ( response ) => { if ( ! response . ok ) { throw new Error ( response . statusText ) ; } return response . body ; } ) . then ( ( rStream ) => { const reader = rStream . getReader ( ) ; const sendChunk = this . sendChunk ; const decoder = this . decoder ; return new ReadableStream ( { start ( ) { let remaining = '' ; function read ( ) { reader . read ( ) . then ( ( { done , value } ) => { if ( done ) { if ( remaining . length ) { sendChunk ( subscriber , remaining ) ; } return subscriber . complete ( ) ; } remaining += decoder ( value ) ; let pos = 0 ; while ( ( pos = remaining . indexOf ( '\n' ) ) > - 1 ) { sendChunk ( subscriber , remaining . slice ( 0 , pos - 1 ) ) ; remaining = remaining . slice ( pos + 1 , remaining . length ) ; } read ( ) ; } ) ; } read ( ) ; } } ) ; } ) . catch ( ( err ) => { subscriber . error ( err ) ; return subscriber . complete ( ) ; } ) ; } ) ; }
Fetches integration data from the server given a search term.
Fetches integration data from the server given a search term .
// nothing to do // NOTE: this assumes that the client has already fetched the integrations // test for bad response code (only on first chunk) // handle each data chunk // stream is done // - 1 = remove the trailing , or ] // keep the rest because it may not be complete // keep reading until done // this catches an issue within the ^ .then