<commit_before>/*
Synctable that uses the project websocket rather than the database.
*/

import { delay } from "awaiting";

import { reuseInFlight } from "async-await-utils/hof";

import { synctable_no_database, SyncTable } from "smc-util/sync/table";

import { once, retry_until_success } from "smc-util/async-utils";

// Always wait at least this long between connect attempts.  This
// avoids flooding the project with connection requests if, e.g., the
// client limit for a particular file is reached.
const MIN_CONNECT_WAIT_MS = 5000;

interface Client {
  touch_project: ({
    project_id,
    cb
  }: {
    project_id: string;
    cb?: Function;
  }) => Promise<any>;
  project_websocket: (project_id: string) => Promise<any>;
  set_connected: (connected: boolean) => void;
}

interface Options {
  project_id: string;
  query: any;
  options: any;
  client: Client;
  throttle_changes?: undefined | number;
  id: string;
}

import { EventEmitter } from "events";

class SyncTableChannel extends EventEmitter {
  public synctable: SyncTable;
  private project_id: string;
  private client: Client;
  private channel?: any;
  private websocket?: any;
  private query: any;
  private options: any;
  private key: string;

  private last_connect: number = 0;

  private connected: boolean = false;

  constructor(opts: Options) {
    super();
    const { project_id, query, options, client, throttle_changes } = opts;
    this.key = key(opts);
    this.synctable = synctable_no_database(
      query,
      options,
      client,
      throttle_changes,
      [],
      project_id
    );
    (this.synctable as any).channel = this; // for debugging
    this.project_id = project_id;
    this.client = client;
    this.query = query;
    this.options = options;
    this.init_synctable_handlers();

    this.connect = reuseInFlight(this.connect.bind(this));
    this.log = this.log.bind(this);
    this.connect();
  }

  public is_connected(): boolean {
    return this.connected;
  }

  private log(..._args): void {
    //console.log("SyncChannel", this.query, ..._args);
  }

  private async connect(): Promise<void> {
    if (this.synctable == null) return;
    this.set_connected(false);
    this.clean_up_sockets();

    const time_since_last_connect = new Date().valueOf() - this.last_connect;
    if (time_since_last_connect < MIN_CONNECT_WAIT_MS) {
      // Last attempt to connect was very recent, so we wait a little before
      // trying again.
      await delay(MIN_CONNECT_WAIT_MS - time_since_last_connect);
    }

    await retry_until_success({
      max_delay: 5000,
      f: this.attempt_to_connect.bind(this),
      desc: "webapp-synctable-connect",
      log: this.log
    });

    this.last_connect = new Date().valueOf();
  }

  private set_connected(connected: boolean): void {
    if (this.synctable == null) return;
    this.log("set_connected", connected);
    this.connected = connected;
    this.synctable.client.set_connected(connected);
    if (connected) {
      this.emit("connected");
    } else {
      this.emit("disconnected");
    }
  }
  // Various things could go wrong, e.g., the websocket breaking
  // while trying to get the api synctable_channel, touch
  // project might time out, etc.
  private async attempt_to_connect(): Promise<void> {
    // Start with fresh websocket and channel -- old one may be dead.
    this.clean_up_sockets();
    // touch_project mainly makes sure that some hub is connected to
    // the project, so the project can do DB queries.  Also
    // starts the project.
    this.client.touch_project({ project_id: this.project_id });
    // Get a websocket.
    this.websocket = await this.client.project_websocket(this.project_id);
    if (this.websocket.state != "online") {
      // give websocket state once chance to change.
      // It could change to destroyed or online.
      await once(this.websocket, "state");
    }
    if (this.websocket.state != "online") {
      // Already offline... let's try again from the top.
      throw Error("websocket went offline already");
    }

    // Get a channel.
    this.channel = await this.websocket.api.synctable_channel(
      this.query,
      this.options
    );

    if (this.websocket.state != "online") {
      // Already offline... let's try again from the top.
      throw Error("websocket went offline already");
    }

    this.channel.on("data", this.handle_mesg_from_project.bind(this));
    this.websocket.on("offline", this.connect);
    this.channel.on("close", this.connect);
  }

  private init_synctable_handlers(): void {
    this.synctable.on("timed-changes", timed_changes => {
      this.send_mesg_to_project({ timed_changes });
    });
    this.synctable.once("closed", this.close.bind(this));
  }

  private clean_up_sockets(): void {
    if (this.channel != null) {
      this.channel.removeListener("close", this.connect);
      try {
        this.channel.end();
      } catch (err) {
        // no op -- this does happen if channel.conn is destroyed
      }
      delete this.channel;
    }

    if (this.websocket != null) {
      this.websocket.removeListener("offline", this.connect);
      delete this.websocket;
    }
  }

  private async close(): Promise<void> {
    delete cache[this.key];
    this.clean_up_sockets();
    if (this.synctable != null) {
      const s = this.synctable;
      delete this.synctable;
      await s.close();
    }
  }

  private handle_mesg_from_project(mesg): void {
    this.log("project --> client: ", mesg);<commit_msg>try to address another problem with primus-multiplex via a stupid hack<commit_after>/*
Synctable that uses the project websocket rather than the database.
*/

import { delay } from "awaiting";

import { reuseInFlight } from "async-await-utils/hof";

import { synctable_no_database, SyncTable } from "smc-util/sync/table";

import { once, retry_until_success } from "smc-util/async-utils";

// Always wait at least this long between connect attempts.  This
// avoids flooding the project with connection requests if, e.g., the
// client limit for a particular file is reached.
const MIN_CONNECT_WAIT_MS = 5000;

interface Client {
  touch_project: ({
    project_id,
    cb
  }: {
    project_id: string;
    cb?: Function;
  }) => Promise<any>;
  project_websocket: (project_id: string) => Promise<any>;
  set_connected: (connected: boolean) => void;
}

interface Options {
  project_id: string;
  query: any;
  options: any;
  client: Client;
  throttle_changes?: undefined | number;
  id: string;
}

import { EventEmitter } from "events";

class SyncTableChannel extends EventEmitter {
  public synctable: SyncTable;
  private project_id: string;
  private client: Client;
  private channel?: any;
  private websocket?: any;
  private query: any;
  private options: any;
  private key: string;

  private last_connect: number = 0;

  private connected: boolean = false;

  constructor(opts: Options) {
    super();
    const { project_id, query, options, client, throttle_changes } = opts;
    this.key = key(opts);
    this.synctable = synctable_no_database(
      query,
      options,
      client,
      throttle_changes,
      [],
      project_id
    );
    (this.synctable as any).channel = this; // for debugging
    this.project_id = project_id;
    this.client = client;
    this.query = query;
    this.options = options;
    this.init_synctable_handlers();

    this.connect = reuseInFlight(this.connect.bind(this));
    this.log = this.log.bind(this);
    this.connect();
  }

  public is_connected(): boolean {
    return this.connected;
  }

  private log(..._args): void {
    //console.log("SyncChannel", this.query, ..._args);
  }

  private async connect(): Promise<void> {
    if (this.synctable == null) return;
    this.set_connected(false);
    this.clean_up_sockets();

    const time_since_last_connect = new Date().valueOf() - this.last_connect;
    if (time_since_last_connect < MIN_CONNECT_WAIT_MS) {
      // Last attempt to connect was very recent, so we wait a little before
      // trying again.
      await delay(MIN_CONNECT_WAIT_MS - time_since_last_connect);
    }

    await retry_until_success({
      max_delay: 5000,
      f: this.attempt_to_connect.bind(this),
      desc: "webapp-synctable-connect",
      log: this.log
    });

    this.last_connect = new Date().valueOf();
  }

  private set_connected(connected: boolean): void {
    if (this.synctable == null) return;
    this.log("set_connected", connected);
    this.connected = connected;
    this.synctable.client.set_connected(connected);
    if (connected) {
      this.emit("connected");
    } else {
      this.emit("disconnected");
    }
  }
  // Various things could go wrong, e.g., the websocket breaking
  // while trying to get the api synctable_channel, touch
  // project might time out, etc.
  private async attempt_to_connect(): Promise<void> {
    // Start with fresh websocket and channel -- old one may be dead.
    this.clean_up_sockets();
    // touch_project mainly makes sure that some hub is connected to
    // the project, so the project can do DB queries.  Also
    // starts the project.
    this.client.touch_project({ project_id: this.project_id });
    // Get a websocket.
    this.websocket = await this.client.project_websocket(this.project_id);
    if (this.websocket.state != "online") {
      // give websocket state once chance to change.
      // It could change to destroyed or online.
      await once(this.websocket, "state");
    }
    if (this.websocket.state != "online") {
      // Already offline... let's try again from the top.
      throw Error("websocket went offline already");
    }

    // Get a channel.
    this.channel = await this.websocket.api.synctable_channel(
      this.query,
      this.options
    );

    if (this.websocket.state != "online") {
      // Already offline... let's try again from the top.
      throw Error("websocket went offline already");
    }

    this.channel.on("data", this.handle_mesg_from_project.bind(this));
    this.websocket.on("offline", this.connect);
    this.channel.on("close", this.connect);
  }

  private init_synctable_handlers(): void {
    this.synctable.on("timed-changes", timed_changes => {
      this.send_mesg_to_project({ timed_changes });
    });
    this.synctable.once("closed", this.close.bind(this));
  }

  private clean_up_sockets(): void {
    if (this.channel != null) {
      this.channel.removeListener("close", this.connect);

      // Explicitly emit end -- this is a hack,
      // since this is the only way to force the
      // channel clean-up code to run in primus-multiplex,
      // and it gets run async later if we don't do this.
      // TODO: rewrite primus-multiplex from scratch.
      this.channel.emit('end');

      try {
        this.channel.end();
      } catch (err) {
        // no op -- this does happen if channel.conn is destroyed
      }
      delete this.channel;
    }

    if (this.websocket != null) {
      this.websocket.removeListener("offline", this.connect);
      delete this.websocket;
    }
  }

  private async close(): Promise<void> {
    delete cache[this.key];
    this.clean_up_sockets();
    if (this.synctable != null) {
      const s = this.synctable;
      delete this.synctable;
      await s.close();
    }
  }

  private handle_mesg_from_project(mesg): void {
    this.log("project --> client: ", mesg);
{'fraction_non_alphanumeric': 0.1202037053297041, 'fraction_numerical': 0.001580472385635262, 'mean_word_length': 2.9617391304347827, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7141479', 'n_tokens_mistral': 3598, 'n_tokens_neox': 3491, 'n_words': 945}