Fuse parallel streams and close
975254
function() {

		// Header length
		var header_length = 32;

		// Write down null-terminated string lookup table in the end
		var stream8offset = this.stream8.offset;
		for (var i=0; i<this.stringLookup.length; ++i) {
			this.stream8.write( new Buffer( this.stringLookup[i] ) );
			this.stream8.write( new Buffer( [0] ) );
		}
		// console.log("ST: len="+(this.stream8.offset - stream8offset)+", sl=", this.stringLookup );

		// Encode the object signature table
		var stream16offset = this.stream16.offset;
		for (var i=0; i<this.plainObjectSignatureTable.length; ++i)
			this.stream16.write( this.plainObjectSignatureTable[i] );
		// console.log("OT: len="+(this.stream16.offset - stream16offset)+", ot=", this.plainObjectSignatureTable );

		// Finalize individual streams
		this.stream64.finalize();
		this.stream32.finalize();
		this.stream16.finalize();
		this.stream8.finalize();

		// Calculate the value to put on the profile table ID
		var profile_id = 0x00;
		if (this.profile.count == 1) {
			profile_id = this.profile.id;
		}

		// If sparse bundle update stream8
		var totalSize = 0;
		if (this.sparse) {

			// Overwrite header
			this.stream8.writeAt( 2,  pack2b( profile_id ) );  	// Update object table ID
			this.stream8.writeAt( 6,  pack2b( this.profile.count ) ); 	// Number of profile tables
			this.stream8.writeAt( 8,  pack4b( this.stream64.offset ) ); // 64-bit buffer lenght
			this.stream8.writeAt( 12, pack4b( this.stream32.offset ) ); // 32-bit buffer lenght
			this.stream8.writeAt( 16, pack4b( this.stream16.offset ) ); // 16-bit buffer length
			this.stream8.writeAt( 20, pack4b( this.stream8.offset - header_length ) );  // 8-bit buffer length (exclude header)
			this.stream8.writeAt( 24, pack4b( this.stream8.offset - stream8offset ) ); // Length of string table
			this.stream8.writeAt( 28, pack4b( this.stream16.offset - stream16offset ) ); // Length of signature table

			// TODO: Separate stream8 in two in order to load them in parallel

			// Close streams
			this.stream8.close();  this.stream16.close(); 
			this.stream32.close(); this.stream64.close(); 

			// Calculate total size
			totalSize = this.stream8.offset +  this.stream16.offset + 
						this.stream32.offset + this.stream64.offset;

			// Update filename
			this.filename += ".jbbp";

			// Write summar header
			if ((this.logFlags & LOG.SUMM)) {
				console.info("Encoding size =",totalSize,"bytes");
				console.info("     8b chunk =",this.stream8.offset,"bytes");
				console.info("    16b chunk =",this.stream16.offset,"bytes");
				console.info("    32b chunk =",this.stream32.offset,"bytes");
				console.info("    64b chunk =",this.stream64.offset,"bytes");

			}

		}

		// If not separating, merge everything
		else {

			// Open final stream
			var finalStream = new BinaryStream( this.filename + '.jbb', 8 );
			finalStream.write( pack2b( 0x4231 ) );  			 // Magic header
			finalStream.write( pack2b( profile_id ) ); 	 		 // Object Table ID
			finalStream.write( pack2b( VERSION ) );	  			 // Version
			finalStream.write( pack2b( this.profile.count ) );   // Number of profile tables
			finalStream.write( pack4b( this.stream64.offset ) ); // 64-bit buffer lenght
			finalStream.write( pack4b( this.stream32.offset ) ); // 32-bit buffer lenght
			finalStream.write( pack4b( this.stream16.offset ) ); // 16-bit buffer length
			finalStream.write( pack4b( this.stream8.offset ) );  // 8-bit buffer length
			finalStream.write( pack4b( this.stream8.offset - stream8offset ) ); // Length of string table
			finalStream.write( pack4b( this.stream16.offset - stream16offset ) ); // Length of signature table

			// Merge individual streams
			finalStream.merge( this.stream64 ); finalStream.merge( this.stream32 );
			finalStream.merge( this.stream16 ); finalStream.merge( this.stream8 );

			// Close
			finalStream.finalize();
			finalStream.close();
			totalSize = finalStream.offset;

			// Close and delete helper stream files
			this.stream8.close();  fs.unlink( this.filename + '.jbbp' );
			this.stream16.close(); fs.unlink( this.filename + '_b16.jbbp' );
			this.stream32.close(); fs.unlink( this.filename + '_b32.jbbp' );
			this.stream64.close(); fs.unlink( this.filename + '_b64.jbbp' );

			// Update filename
			this.filename += ".jbb";

			// Write summar header
			if ((this.logFlags & LOG.SUMM)) {
				console.info("Encoding size =",finalStream.offset,"bytes");
			}

		}

		// Write protocol overhead table
		if ((this.logFlags & LOG.PDBG) != 0) {
			console.info("-----------------------------------");
			console.info(" Binary Protocol Overhead Analysis");
			console.info("-----------------------------------");
			console.info(" 64-bit Stream       : ", this.stream64.offset, "b");
			console.info(" 32-bit Stream       : ", this.stream32.offset, "b");
			console.info(" 16-bit Stream       : ", this.stream16.offset, "b");
			console.info("  8-bit Stream       : ", this.stream8.offset,  "b");
			console.info("-----------------------------------");
			console.info(" Control Op-Codes    : ", this.counters.op_ctr, "b");
			console.info(" Primitive Op-Codes  : ", this.counters.op_prm, "b");
			console.info(" String References   : ", this.counters.ref_str, "b");
			console.info(" I-Refs              : ", this.counters.op_iref, "b");
			console.info(" X-Refs              : ", this.counters.op_xref, "b");
			console.info(" Array Headers       : ", this.counters.arr_hdr, "b");
			console.info(" Array Steering Data : ", this.counters.arr_dat, "b");
			console.info(" Array Chunks        : ", this.counters.arr_chu, "b");
			console.info(" Buffer Headers      : ", this.counters.dat_hdr, "b");
			var sum = this.counters.op_ctr + this.counters.op_prm + this.counters.arr_dat +
					  this.counters.ref_str + this.counters.op_iref + this.counters.op_xref +
					  this.counters.arr_hdr + this.counters.arr_chu + this.counters.dat_hdr;
			console.info("-----------------------------------");
			console.info(" NUM_DWS             : ",this.counters.arr_dws);
			console.info(" NUM_DELTA_FLOAT     : ",this.counters.arr_delta_float);
			console.info(" NUM_DELTA_INT       : ",this.counters.arr_delta_int);
			console.info(" NUM_REPEATED        : ",this.counters.arr_num_repeated);
			console.info(" NUM_RAW             : ",this.counters.arr_num_raw );
			console.info(" NUM_SHORT           : ",this.counters.arr_num_short);
			console.info(" PRIM_BULK_PLAIN     : ",this.counters.arr_prim_bulk_plain);
			console.info(" PRIM_BULK_KNOWN     : ",this.counters.arr_prim_bulk_known);
			console.info(" PRIM_SHORT          : ",this.counters.arr_prim_short);
			console.info(" PRIM_REPEATED       : ",this.counters.arr_prim_repeated);
			console.info(" PRIM_RAW            : ",this.counters.arr_prim_raw);
			console.info(" PRIM_CHUNK          : ",this.counters.arr_prim_chunk);
			console.info(" EMPTY               : ",this.counters.arr_empty);
			console.info("-----------------------------------");
			console.info("");
			var perc = ((sum / totalSize) * 100).toFixed(2);
			console.info(" Total Overhead     : ",sum,"b ("+perc+" %)");
		}

	}
715296
public void close() throws IOException, InterruptedException {
        channel.close();
        channel.join();
        if (ownsPool)
            pool.shutdown();
        for (Closeable c : closables)
            c.close();
    }
610898
def end_parallel(self):
        
        outport = self.oport
        if isinstance(self.oport.operator, streamsx.topology.graph.Marker):
            if self.oport.operator.kind == "$Union$":
                pto = self.topology.graph.addPassThruOperator()
                pto.addInputPort(outputPort=self.oport)
                outport = pto.addOutputPort(schema=self.oport.schema)
        op = self.topology.graph.addOperator("$EndParallel$")
        op.addInputPort(outputPort=outport)
        oport = op.addOutputPort(schema=self.oport.schema)
        endP = Stream(self.topology, oport)
        return endP
873093
function concat(divider, sourcePath, relativePaths, done) {
  var writeDivider = false;
  each(relativePaths, function(relativePath, next) {
    var absoluteSourcePath = path.resolve(sourcePath, relativePath);
    var readStream = fs.createReadStream(absoluteSourcePath);
    if (writeDivider) process.stdout.write(divider);
    writeDivider = true;
    readStream.on('close', next).on('error', next).pipe(process.stdout);
  }, done);
}
237961
void close() {
    shuttingDown = true;
    // shutdown thread pools
    dispatcherExecutor.shutdownNow();
    moverExecutor.shutdownNow();
    // close the output file
    IOUtils.closeStream(out);
    if (fs != null) {
      try {
        fs.delete(BALANCER_ID_PATH, true);
      } catch(IOException ignored) {
      }
    }
  }
218166
public static void closeStreams(Process self) {
        try { self.getErrorStream().close(); } catch (IOException ignore) {}
        try { self.getInputStream().close(); } catch (IOException ignore) {}
        try { self.getOutputStream().close(); } catch (IOException ignore) {}
    }
485483
@Override
  public final void close()
  {
    try {
      TempBufferData tempBuffer = _tempRead;
      
      if (tempBuffer != null && ! _isReuseBuffer) {
        _tempRead = null;
        _readBuffer = null;

        tempBuffer.free();
      }

      if (_source != null) {
        StreamImpl s = _source;
        _source = null;
        s.close();
      }
    } catch (IOException e) {
      log.log(Level.FINE, e.toString(), e);
    }
  }
30572
void encodeStripe(
    InputStream[] blocks,
    long blockSize,
    OutputStream[] outs,
    CRC32[] crcOuts,
    Progressable reporter,
    boolean computeSrcChecksum,
    List<Integer> errorLocations) throws IOException {
    configureBuffers(blockSize);
    int boundedBufferCapacity = 1;
    ParallelStreamReader parallelReader = new ParallelStreamReader(
      reporter, blocks, bufSize, 
      parallelism, boundedBufferCapacity, blockSize, computeSrcChecksum,
      outs);
    parallelReader.start();
    try {
      for (long encoded = 0; encoded < blockSize; encoded += bufSize) {
        ParallelStreamReader.ReadResult readResult = null;
        try {
          readResult = parallelReader.getReadResult();
        } catch (InterruptedException e) {
          throw new IOException("Interrupted while waiting for read result");
        }
        // Cannot tolerate any IO errors.
        IOException readEx = readResult.getException();
        if (readEx != null) {
          if (errorLocations != null) {
            errorLocations.clear();
            for (int idx : readResult.getErrorIdx()) {
              errorLocations.add(idx);
            }
          }
          throw readEx;
        }

        code.encodeBulk(readResult.readBufs, writeBufs);
        reporter.progress();
        // Assume each byte is independently encoded
        int toWrite = (int)Math.min(blockSize - encoded, bufSize);

        // Now that we have some data to write, send it to the temp files.
        for (int i = 0; i < codec.parityLength; i++) {
          outs[i].write(writeBufs[i], 0, toWrite);
          if (crcOuts != null && crcOuts[i] != null) {
            crcOuts[i].update(writeBufs[i], 0, toWrite);
          }
          reporter.progress();
        }
      }
      if (computeSrcChecksum) {
        parallelReader.collectSrcBlocksChecksum(checksumStore);
      }
    } finally {
      parallelReader.shutdown();
    }
  }
1343725
@Override
  public void close()
    throws IOException
  {
    if (_isClosed.getAndSet(true)) {
      return;
    }

    _ss = null;

    if (_stream != null) {
      _stream.close();
    }

    // XXX: can't be locked because of shutdown

    _nativeFd = -1;
    nativeClose(_socketFd);
  }
1255170
void free(ClientSocket stream)
  {
    success();

    _activeCount.decrementAndGet();

    synchronized (this) {
      int size = (_idleHead - _idleTail + _idle.length) % _idle.length;

      if (_state != State.CLOSED && size < _idleSize) {
        _idleHead = (_idleHead + 1) % _idle.length;
        _idle[_idleHead] = stream;

        stream = null;
      }

      long now = CurrentTime.currentTime();

      long prevSuccessTime = _prevSuccessTime;

      if (prevSuccessTime > 0) {
        _latencyFactor = (0.95 * _latencyFactor
                          + 0.05 * (now - prevSuccessTime));
      }

      if (_activeCount.get() > 0)
        _prevSuccessTime = now;
      else
        _prevSuccessTime = 0;

      _lastSuccessTime = now;
      
      if (log.isLoggable(Level.FINEST)) { 
        logFinest(L.l("free: _lastSuccessTime={0}, _failTime={1}",
                      now, _failTime));
      }
    }

    updateWarmup();

    long now = CurrentTime.currentTime();
    long maxIdleTime = _loadBalanceIdleTime;
    ClientSocket oldStream = null;

    do {
      oldStream = null;

      synchronized (this) {
        if (_idleHead != _idleTail) {
          int nextTail = (_idleTail + 1) % _idle.length;

          oldStream = _idle[nextTail];

          if (oldStream != null
              && oldStream.getIdleStartTime() + maxIdleTime < now) {
            _idle[nextTail] = null;
            _idleTail = nextTail;
          }
          else
            oldStream = null;
        }
      }

      if (oldStream != null)
        oldStream.closeImpl();
    } while (oldStream != null);

    if (stream != null) {
      stream.closeImpl();
    }
  }
1271140
def bind_end(self, stream, future):
		""""""
		if not isinstance(stream, Stream):
			future.set_result(None)
		else:
			stream.pipe(TaskEndTransformer(future))
1366588
@Override
  public void closeWrite() throws IOException
  {
    if (_isWriteClosed) {
      return;
    }
    
    _isWriteClosed = true;
    
    SocketStream stream = _streamImpl;
    
    if (stream != null) {
      stream.closeWrite();
    }
    else if (_s != null) {
      try {
        _s.shutdownOutput();
      } catch (UnsupportedOperationException e) {
        log.log(Level.FINEST, e.toString(), e);
      } catch (Exception e) {
        log.finer(e.toString());
        log.log(Level.FINEST, e.toString(), e);
      }
    }
  }
1162591
function close() {
		if ( error ) {
			debug( 'Stream was destroyed due to an error. Error: %s.', JSON.stringify( error ) );
			self.emit( 'error', error );
		}
		debug( 'Closing the stream...' );
		self.emit( 'close' );
	}
1105237
synchronized void closeChunk( int cidx, int len ) {
    // The Parser will pre-allocate the _tmp_espc large enough (the Parser
    // knows how many final Chunks there will be up front).  Other users are
    // encouraged to set a "large enough" espc - and a shared one at that - to
    // avoid these copies.

    // Set the length into the temp ESPC at the Chunk index (accounting for _chunkOff)
    cidx -= _chunkOff;
    while( cidx >= _tmp_espc.length ) // should not happen if espcs are preallocated and shared!
      _tmp_espc = Arrays.copyOf(_tmp_espc, _tmp_espc.length<<1);
    _tmp_espc[cidx] = len;
  }
94659
private static void closeStreams( InputStream src, OutputStream dest )
    {
        try
        {
            src.close();
        }
        catch( IOException e )
        {
            e.printStackTrace();
        }

        try
        {
            dest.close();
        }
        catch( IOException e )
        {
            e.printStackTrace();
        }
    }
35471
protected void implCloseChannel() throws IOException {
        // Release and invalidate any locks that we still hold
        if (fileLockTable != null) {
            for (FileLock fl: fileLockTable.removeAll()) {
                synchronized (fl) {
                    if (fl.isValid()) {
                        nd.release(fd, fl.position(), fl.size());
                        ((FileLockImpl)fl).invalidate();
                    }
                }
            }
        }

        threads.signalAndWait();

        if (parent != null) {

            // Close the fd via the parent stream's close method.  The parent
            // will reinvoke our close method, which is defined in the
            // superclass AbstractInterruptibleChannel, but the isOpen logic in
            // that method will prevent this method from being reinvoked.
            //
            ((java.io.Closeable)parent).close();
        } else {
            nd.close(fd);
        }

    }
1285784
public void close() {
    proc.destroy();
    try {
      inStream.close();
      outStream.close();
      errStream.close();
    } catch (IOException ignored) {
    }
    inStream = null;
    outStream = null;
    errStream = null;
    proc = null;
  }