<commit_before>import * as pathToRegexp from 'path-to-regexp';
import { Readable } from 'stream';
import { Rowan, IRowan, Handler } from 'rowan';

import { Events } from './utils/events';
import { Wire } from './wire';
import { Method } from './method';
import { Context } from './context';
import { Message, MessageFlags } from './message';
import { MessageSerialiser } from './message-serialiser';

import { ViaPath, ViaHandler, ViaInterceptor } from './via-types';
export { ViaPath, ViaHandler, ViaInterceptor } from './via-types';

import { unhandled, unhandledError, fatalError } from './middleware/unhandled';

export class Via implements IRowan<Context> {
  private _root = new Rowan<Context>();
  private _app = new Rowan<Context>();

  protected _serialiser = new MessageSerialiser();
  protected _interceptors = new Map<string, ViaInterceptor>();

  constructor(private wire?: Wire) {
    //TODO: Move these out;     
    this._root.use(this.interceptors());
    this._root.use(this.responseStreamer());

    this._root.use(this._app);

    // Fallbacks 
    this._root.use(unhandled());
    this._root.use(unhandledError());
    this._root.use(fatalError());

    if (!!wire) {
      wire.on("message", (data) => this.decodeWireMessage(wire, new Uint8Array(data)));
    }
  }
  private intercept(id: string, handler: ViaHandler, ...handlers: ViaHandler[]): () => void {
    let dispose = () => this._interceptors.delete(id);

    this._interceptors.set(id, {
      dispose: dispose,
      handlers: [handler, ...handlers]
    });

    return dispose;
  }

  use(handler: ViaHandler, ...handlers: ViaHandler[]) {
    this._app.use(handler, ...handlers);
    return this;
  }

  get(path: ViaPath, handler: ViaHandler, ...handlers: ViaHandler[]) {
    this.method(Method.GET, path, handler, ...handlers);
    return this;
  }

  put(path: ViaPath, handler: ViaHandler, ...handlers: ViaHandler[]) {
    this.method(Method.PUT, path, handler, ...handlers);
    return this;
  }

  post(path: ViaPath, handler: ViaHandler, ...handlers: ViaHandler[]) {
    this.method(Method.POST, path, handler, ...handlers);
    return this;
  }

  patch(path: ViaPath, handler: ViaHandler, ...handlers: ViaHandler[]) {
    this.method(Method.PATCH, path, handler, ...handlers);
    return this;
  }

  delete(path: ViaPath, handler: ViaHandler, ...handlers: ViaHandler[]) {
    this.method(Method.DELETE, path, handler, ...handlers);
    return this;
  }

  subscribe(path: ViaPath, handler: ViaHandler, ...handlers: ViaHandler[]) {
    this.method(Method.SUBSCRIBE, path, handler, ...handlers);
    return this;
  }

  unsubscribe(path: ViaPath, handler: ViaHandler, ...handlers: ViaHandler[]) {
    this.method(Method.UNSUBSCRIBE, path, handler, ...handlers);
    return this;
  }

  method(method: Method, path: ViaPath, handler: ViaHandler, ...handlers: ViaHandler[]) {
    this.use(
      (ctx) => ctx.req.method === method,
      this.pathHandler(path),
      handler,
      ...handlers);
    return this;
  }

  path(path: ViaPath, handler: ViaHandler, ...handlers: ViaHandler[]) {
    this.use(this.pathHandler(path), handler, ...handlers);
    return this;
  }

  private pathHandler(path: ViaPath) {
    let keys = [];
    var exp = pathToRegexp(path, keys);
    return async (ctx: Context) => {
      let match = (ctx.req.path) ? exp.exec(ctx.req.path) : null;
      if (match == null) {
        return false;
      }
      if (keys.length > 0) {
        ctx.req.params = ctx.req.params || {};
        for (let i = 0; i < keys.length; i += 1) {
          ctx.req.params[keys[i].name] = match[i + 1];
        }
      }
    };
  }

  /* solicited or unsolicited response */
  send(msg: Message, ...wires: Wire[]) {
    wires = wires || [this.wire];

    if (wires == undefined)
      throw Error("wire is undefined");

    let bin = this._serialiser.encode(msg).buffer;

    for (var wire of wires){
      wire.send(bin);
    }
  }

  request(msg: Message, ...handlers: ViaHandler[]) {
    if (msg.id == undefined)
      msg.id = Message.genIdString();

    let bin = this._serialiser.encode(msg).buffer;

    var resolve;
    let promise = new Promise<Message>((r) => resolve = r);

    var dispose = this.intercept(msg.id, (ctx) => {
      resolve(ctx);
      dispose();
    }, ...handlers);

    this.wire.send(bin);
    return promise;
  }

  process(err: any, ctx: Context) {
    if (arguments.length == 2)
      return this._root.process(err, ctx);
    return this._root.process(ctx);
  }

  protected async decodeWireMessage(wire: Wire, binary: Uint8Array) {
    let msg = this._serialiser.decode(binary);

    let ctx = this.createContext(wire, msg);

    try {
      await this.process(undefined, ctx);
    }
    catch (err) {
      console.log("process exception:", err);
    }
  }

  private createContext(wire: Wire, msg: Message) {
    let ctx: Context = {
      wire: wire,
    };

    if (msg.status != undefined) {
      ctx.res = msg;
      return ctx;
    }

    ctx.req = msg;
    ctx.res = { id: msg.id };
    ctx.begin = () => {
      let sid = Message.genIdString();<commit_msg>fix: overloading of process method<commit_after>import * as pathToRegexp from 'path-to-regexp';
import { Readable } from 'stream';
import { Rowan, IRowan, Handler } from 'rowan';

import { Events } from './utils/events';
import { Wire } from './wire';
import { Method } from './method';
import { Context } from './context';
import { Message, MessageFlags } from './message';
import { MessageSerialiser } from './message-serialiser';

import { ViaPath, ViaHandler, ViaInterceptor } from './via-types';
export { ViaPath, ViaHandler, ViaInterceptor } from './via-types';

import { unhandled, unhandledError, fatalError } from './middleware/unhandled';

export class Via implements IRowan<Context> {
  private _root = new Rowan<Context>();
  private _app = new Rowan<Context>();

  protected _serialiser = new MessageSerialiser();
  protected _interceptors = new Map<string, ViaInterceptor>();

  constructor(private wire?: Wire) {
    //TODO: Move these out;     
    this._root.use(this.interceptors());
    this._root.use(this.responseStreamer());

    this._root.use(this._app);

    // Fallbacks 
    this._root.use(unhandled());
    this._root.use(unhandledError());
    this._root.use(fatalError());

    if (!!wire) {
      wire.on("message", (data) => this.decodeWireMessage(wire, new Uint8Array(data)));
    }
  }
  private intercept(id: string, handler: ViaHandler, ...handlers: ViaHandler[]): () => void {
    let dispose = () => this._interceptors.delete(id);

    this._interceptors.set(id, {
      dispose: dispose,
      handlers: [handler, ...handlers]
    });

    return dispose;
  }

  use(handler: ViaHandler, ...handlers: ViaHandler[]) {
    this._app.use(handler, ...handlers);
    return this;
  }

  get(path: ViaPath, handler: ViaHandler, ...handlers: ViaHandler[]) {
    this.method(Method.GET, path, handler, ...handlers);
    return this;
  }

  put(path: ViaPath, handler: ViaHandler, ...handlers: ViaHandler[]) {
    this.method(Method.PUT, path, handler, ...handlers);
    return this;
  }

  post(path: ViaPath, handler: ViaHandler, ...handlers: ViaHandler[]) {
    this.method(Method.POST, path, handler, ...handlers);
    return this;
  }

  patch(path: ViaPath, handler: ViaHandler, ...handlers: ViaHandler[]) {
    this.method(Method.PATCH, path, handler, ...handlers);
    return this;
  }

  delete(path: ViaPath, handler: ViaHandler, ...handlers: ViaHandler[]) {
    this.method(Method.DELETE, path, handler, ...handlers);
    return this;
  }

  subscribe(path: ViaPath, handler: ViaHandler, ...handlers: ViaHandler[]) {
    this.method(Method.SUBSCRIBE, path, handler, ...handlers);
    return this;
  }

  unsubscribe(path: ViaPath, handler: ViaHandler, ...handlers: ViaHandler[]) {
    this.method(Method.UNSUBSCRIBE, path, handler, ...handlers);
    return this;
  }

  method(method: Method, path: ViaPath, handler: ViaHandler, ...handlers: ViaHandler[]) {
    this.use(
      (ctx) => ctx.req.method === method,
      this.pathHandler(path),
      handler,
      ...handlers);
    return this;
  }

  path(path: ViaPath, handler: ViaHandler, ...handlers: ViaHandler[]) {
    this.use(this.pathHandler(path), handler, ...handlers);
    return this;
  }

  private pathHandler(path: ViaPath) {
    let keys = [];
    var exp = pathToRegexp(path, keys);
    return async (ctx: Context) => {
      let match = (ctx.req.path) ? exp.exec(ctx.req.path) : null;
      if (match == null) {
        return false;
      }
      if (keys.length > 0) {
        ctx.req.params = ctx.req.params || {};
        for (let i = 0; i < keys.length; i += 1) {
          ctx.req.params[keys[i].name] = match[i + 1];
        }
      }
    };
  }

  /* solicited or unsolicited response */
  send(msg: Message, ...wires: Wire[]) {
    wires = wires || [this.wire];

    if (wires == undefined)
      throw Error("wire is undefined");

    let bin = this._serialiser.encode(msg).buffer;

    for (var wire of wires){
      wire.send(bin);
    }
  }

  request(msg: Message, ...handlers: ViaHandler[]) {
    if (msg.id == undefined)
      msg.id = Message.genIdString();

    let bin = this._serialiser.encode(msg).buffer;

    var resolve;
    let promise = new Promise<Message>((r) => resolve = r);

    var dispose = this.intercept(msg.id, (ctx) => {
      resolve(ctx);
      dispose();
    }, ...handlers);

    this.wire.send(bin);
    return promise;
  }

  process(ctx: Context)
  process(err: any, ctx: Context)
  process() {
    if (arguments.length == 2)
      return this._root.process(arguments[0], arguments[1]);

    return this._root.process(arguments[0]);
  }


  protected async decodeWireMessage(wire: Wire, binary: Uint8Array) {
    let msg = this._serialiser.decode(binary);

    let ctx = this.createContext(wire, msg);

    try {
      await this.process(undefined, ctx);
    }
    catch (err) {
      console.log("process exception:", err);
    }
  }

  private createContext(wire: Wire, msg: Message) {
    let ctx: Context = {
      wire: wire,
    };

    if (msg.status != undefined) {
      ctx.res = msg;
      return ctx;
    }

    ctx.req = msg;
    ctx.res = { id: msg.id };
    ctx.begin = () => {
      let sid = Message.genIdString();
{'fraction_non_alphanumeric': 0.16115863141524106, 'fraction_numerical': 0.0016524105754276827, 'mean_word_length': 2.9211128048780486, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 31, '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': '1614509', 'n_tokens_mistral': 3447, 'n_tokens_neox': 3250, 'n_words': 798}