Cyphon

Search

Overview

The Search class orchestrates search operations across multiple sources. It uses a message queue for managing search jobs, processes these jobs asynchronously, and collects results from various sources. This is particularly useful for scenarios requiring distributed or multi-source search operations.  

\n

Code Details

Properties

public mqqueue: MQQueue; 
private searchQueue;
private sources: BaseSearch<any>[];
private parser = new ExpressionParser();
private capabilities: { [key: string]: Array<number> } = {};
  • mqqueue: Manages search jobs in a message queue.
  • searchQueue: Handles execution of queued jobs.
  • sources: List of search sources to query.
  • parser: Parses search query expressions.
  • capabilities: Maps search source capabilities to their indices.

Constructor

constructor(sources: BaseSearch<any>[]) {

The constructor initializes the Search class:

  1. Accepts an array of BaseSearch sources.
  2. Initializes the MQQueue instance.
  3. Creates a worker for the searchQueue.
  4. Maps the capabilities of each search source.

Public Methods

runSearch

public async runSearch(searchRequest: SearchRequest): Promise<string> {
  • Description: Adds a search request to the queue and returns the job ID.
  • Reasoning: By queuing the search request, it allows the system to handle multiple jobs concurrently, improving scalability and responsiveness.
  • Parameters:
    • searchRequest: Contains the search query and parameters.
  • Returns: A promise resolving to the job ID.

runSearchSync

public async runSearchSync(searchRequest: SearchRequest): Promise<any> {
  • Description: Executes the search request synchronously and returns the results.
  • Reasoning: Useful for scenarios where the caller needs immediate results and can tolerate potential delays.
  • Parameters:
    • searchRequest: Contains the search query and parameters.
  • Returns: The search results.

getJobData

public async getJobData(jobId: string): Promise<any | null> {
  • Description: Retrieves the status and data of a search job.
  • Reasoning: Provides transparency and traceability by allowing the caller to monitor job progress and fetch results.
  • Parameters:
    • jobId: The ID of the job to fetch.
  • Returns: Job data including results, progress, and completion status.

Private Methods

collectDataFromSource

private async collectDataFromSource<T>(...): Promise<void> {
  • Description: Collects search data from a single source, with a timeout mechanism.
  • Reasoning: Using a timeout ensures that unresponsive or slow sources do not block the entire search process. This enhances system reliability and ensures predictable execution times.
  • Parameters:
    • source: The search source to query.
    • searchRequest: The request to process.
    • collectedData: Array to store results.
    • job: The job object to update progress and data.
    • updateProgress: Callback to update job progress.
    • timeout: Timeout duration (default: 60 seconds).

withTimeout

private async withTimeout(promise: Promise<any>, timeout: number): Promise<any> {
  • Description: Wraps a promise with a timeout, rejecting it if the timeout is reached.
  • Reasoning: Adds a safeguard against indefinite execution by imposing a maximum runtime for operations, ensuring system responsiveness.
  • Parameters:
    • promise: The promise to wrap.
    • timeout: Timeout duration.
  • Returns: The resolved promise value or rejects on timeout.

getCollectors

private getCollectors(searchRequest, collectedData, job, updateProgress) {
  • Description: Prepares data collectors for each search source based on their capabilities.
  • Reasoning: By filtering and preparing requests specific to source capabilities, this method ensures that each source receives a compatible query, optimizing resource usage and accuracy.
  • Parameters:
    • searchRequest: The original search request.
    • collectedData: Array to store results.
    • job: The job object to update progress and data.
    • updateProgress: Callback to update progress.
  • Returns: Array of promises for each source's data collection process.

processSearchJob

private async processSearchJob(job): Promise<{ results: any }> {
  • Description: Processes a search job, coordinating data collection from all sources.
  • Reasoning: Centralizes the search execution logic, ensuring consistent handling of requests and aggregation of results from diverse sources.
  • Parameters:
    • job: The search job to process.
  • Returns: A promise resolving to the aggregated search results.

Workflow

  1. Job Initialization:
    • A search request is created and parsed using runSearch or runSearchSync.
    • The parsed query is passed to the appropriate sources based on their capabilities.
  2. Data Collection:
    • collectDataFromSource handles data collection for each source asynchronously.
    • Progress is updated via updateProgress.
  3. Timeout Management:
    • Promises are wrapped in a timeout using withTimeout to ensure timely execution.
  4. Result Aggregation:
    • Collected data is aggregated and returned as the final result.

Example Usage

const sources = [new SourceA(), new SourceB()];
const search = new Search(sources);

const searchRequest = {
  query: "methodA(arg1, arg2) and methodB(arg3, arg4)",
  owner: "user123",
};

// Run asynchronously
const jobId = await search.runSearch(searchRequest);
console.log(`Job ID: ${jobId}`);

// Get job data
const jobData = await search.getJobData(jobId);
console.log(jobData);

// Run synchronously
const results = await search.runSearchSync(searchRequest);
console.log(results);