public _start(out: Stream<Array<R>>): void {
    this.out = out;
    this.ins._add(this);
  }