/* AmqpStream ( options, callback )
3508
function ( options, callback ) {
    
    if ( typeof options == 'function' && typeof callback != 'function' ) {
        callback = options;
    }
    var callback = callback || function (){};

    // Validate arguments
    if ( typeof options != 'object' || options.constructor != Object ) {
        callback( Error( 'AmqpStream requires the first argument to an options object.' ) );
        return false;
    }

    // If module was called as a function, return a new AmqpStream object
    if ( !(this instanceof AmqpStream) ) 
        return new AmqpStream( options, callback );

    // Variables / Properties
    var stream          = this;
    var err             = null;

    // Make sure we have a valid connection
    if ( !(amqpObjectType(options.connection)=='Connection') && !(amqpObjectType(options.exchange)=='Exchange') && !(amqpObjectType(options.queue)=='Queue') ) {
        callback( Error( 'AmqpStream requires at least one amqp object (Connection, Exchange, or Queue).' ) );
        return false;
    }

    // Determine connection bearing object
    var connection = amqpObjectType(options.connection)=='Connection' ? options.connection : 
        amqpObjectType(options.exchange)=='Exchange' ? options.exchange.connection :
        amqpObjectType(options.queue)=='Queue' ? options.queue.connection : null;

    // AMQP related configuration properties
    this.connection     = null;
    this.exchange       = null;
    this.queue          = null;
    this.routingKey     = options.routingKey || '#';
    this.ctag           = null;
    this.originId       = uniqueIdentifier();
    this.autoBind       = typeof options.autoBind != 'undefined' ? options.autoBind : true;

    this.correlations   = [];
    //this.correlationId  = (options.correlationId && options.correlationId === true) ? uniqueIdentifier() : (options.correlationId || null);

    this.publishOptions         = options.publishOptions || {};
    this.publishOptions.headers = this.publishOptions.headers || {};
    this.publishOptions.headers['x-stream-origin'] = this.originId;

    // Stream interface properties
    this.writable       = false;    // Enabled once exchange is set
    this.readable       = false;
    this.paused         = false;    // Start in an un-paused state
    this.ended          = false;
    this.destroyed      = false;
    this.buffer         = [];

    this.initializing   = true;

    var connectionReady = false;
    var exchangeReady   = false;
    var queueReady      = false;
    var onReady = function () {
        if ( stream.initializing && connectionReady && (exchangeReady || !options.exchange) && (queueReady || (typeof options.queue != 'undefined' && options.queue === false) ) ) {
            stream.initializing = false;
            callback( err, stream );
            process.nextTick( function () { stream.emit( 'ready' ); } );
        }
    }

    var configureStream = function ( connection, options ) {
        var connectionReady = function () {
            setConnection( connection );

            if ( amqpObjectType(options.exchange) == 'Exchange' ) setExchange( options.exchange );
            else if ( options.exchange && typeof options.exchange == 'string' ) {
                connection.exchange( options.exchange, options.exchangeOptions||{durable:false,autoDelete:true}, function (exchange) {
                    setExchange( exchange );
                });
            } 

            if ( amqpObjectType(options.queue) == 'Queue' ) { 
                if ( !options.exchange && options.queue.exchange ) {
                    options.exchange = options.queue.exchange;
                    setExchange( options.exchange );
                }
                setQueue( options.queue );
            }
            else if ( options.queue && typeof options.queue == 'string' ) {
                connection.queue( options.queue, options.queueOptions||{durable:false,autoDelete:true}, function (queue) {
                    setQueue( queue );
                });
            }
            // Skip binding to a queue if queue:false, will create write-only stream 
            else if ( typeof options.queue != 'undefined' && options.queue === false ) {
                options.queue = null;
            }
            else {
                options.queue = buildQueue();
            }

        }
        if ( !connection._connecting && connection.channels && Object.keys(connection.channels).length > 0 ) {
            connectionReady();
        }
        else {
            connection.on( 'ready', connectionReady );
        }
    }

    var setConnection = function ( connection ) {
        stream.connection = connection;
        stream.emit( 'connectionReady' );

        connectionReady = true;
        onReady();
    }
    var setExchange = function ( exchange ) {
        if ( stream.exchange )
            return;
        assert.equal( amqpObjectType(exchange), 'Exchange', 'setExchange expected an Exchange object' );
        stream.exchange = exchange;
        stream.writable = true;
        stream.emit( 'exchangeReady' );
        stream.exchange.on( 'error', function (err) { stream.emit( 'error', err); });

        exchangeReady = true;
        onReady();
    }
    var setQueue = function ( queue ) {
        if ( stream.queue ) 
            return;
        assert.equal( amqpObjectType(queue), 'Queue', 'setQueue expected a Queue object' );
        stream.queue = queue;
        stream.readable = true;
        stream.emit( 'queueReady' );
        queue.on( 'error', function (err) { stream.emit( 'error', err); });
        queue.subscribe( messageHandler ).addCallback( function(ok) { 
            stream.ctag = ok.consumerTag;
        });

        // if ( stream.correlationId ) {
        //     stream.publishOptions.replyTo = queue.name;
        //     stream.publishOptions.correlationId = stream.correlationId;
        // }

        queueReady = true;
        onReady();
    }
    var buildQueue = function ( ) {
        var queue = stream.connection.queue( '', {durable: false, exclusive: true, autoDelete: true}, function () {
            if ( stream.autoBind )
                queue.bind( stream.exchange, stream.routingKey );
            queue.bind( stream.exchange, queue.name );
            queue.on( 'queueBindOk', function () { setQueue(queue); });
        });
        return queue;
    }
    var onEnd = function () {
        stream.writable = stream.readable  = false;
        stream.ended = true;
        stream.emit( 'end' );
    }
    var messageHandler = function ( message, headers, deliveryInfo ) {
        if ( 'x-stream-origin' in headers && headers['x-stream-origin'] == stream.originId )
            return; // Do not emit our own writes ( this is for duplex streams )
        if ( deliveryInfo.correlationId ) {
            if ( deliveryInfo.correlationId in stream.correlations ) {
                var emitter = stream.correlations[ deliveryInfo.correlationId ];
            }
            else {
                var emitter = stream.createCorrelatedResponse( deliveryInfo.replyTo, deliveryInfo.correlationId, function ( err, emitter ) {
                    stream.emit( 'correlatedRequest', emitter );
                });
            }
            if ( 'x-stream-event' in headers && headers['x-stream-event'] == 'end') {
                if ( message.data && message.data.toString().trim() != '' )
                    emitter.emit( 'data', message.data );
                emitter.emit( 'end' );
                process.nextTick( function () { delete stream.correlations[deliveryInfo.correlationId] } );
            }
        }
        else {
            var emitter = stream;
            if ( 'x-stream-event' in headers && headers['x-stream-event'] == 'end' ) {

                if ( message.data && message.data.toString().trim() != '' )
                    emitter.emit( 'data', message.data );

                if( !( emitter.paused || emitter.buffer.length ) )
                    return onEnd();
                else
                    emitter.once( 'drain', onEnd );
                emitter.drain();
            }
        }
        
        if ( 'x-stream-event' in headers && headers['x-stream-event'] == 'error') {
            if ( message.data && message.data.toString().trim() != '' )
                emitter.emit( 'data', message.data );
            emitter.emit( 'error' );
        }
        else if ( ( !('x-stream-event' in headers) || ('x-stream-event' in headers && headers['x-stream-event'] == 'data') ) && 
            message.data && message.data.toString().trim() != '' 
        ) {
            if ( emitter.paused ) {
                emitter.buffer.push( message.data );
            }
            else {
                emitter.emit( 'data', message.data );
            }
        }
    }
    
    // Configure the stream
    configureStream( connection, options );
}
448823
function(options) {
    if (typeof options === 'undefined') {
      throw new Error('new StreamChannels(options) - options parameter missing');
    }
    else if (typeof options.track === 'undefined') {
      throw new Error('new StreamChannels(options) - options.track parameter missing');
    }
    else if (!(typeof options.track === 'object' || typeof options.track === 'string')) {
      throw new Error('new StreamChannels(options) - options.track must be an Object (representing your channels with there keywords), an Array (of keywords) or a String (with comma separeted keywords)');
    }
  }
258935
function (options) {
    this.queue = [];
    this.ownerTokenId = null;
    this.options = _.extend({}, AsyncLock.defaultOptions, options);
}
432109
function(dbOrUri, stream, callback) {
    var db;
    if (arguments.length === 2) {
      callback = stream;
      stream = undefined;
    }
    if (!stream) {
      stream = process.stdin;
    }

    var state = 'start';
    var version;
    var collection;
    var endOfCollection;
    var insertQueue = [];
    var insertQueueSize = 0;

    // State machine

    var processors = {
      start: function(item, callback) {
        try {
          item = BSON.deserialize(item);
        } catch (e) {
          return callback(e);
        }
        if (typeof(item) !== 'object') {
          return callback(new Error('First BSON element is not an object'));
        }
        if (item.type !== 'mongo-dump-stream') {
          return callback(new Error('type property is not mongo-dump-stream'));
        }
        version = item.version;
        if (item.version > 2) {
          return callback(new Error('Incoming mongo-dump-stream is of a newer version, bailing out'));
        }
        endOfCollection = item.endOfCollection;
        state = 'collection';
        return callback(null);
      },
      collection: function(item, callback) {
        try {
          item = BSON.deserialize(item);
        } catch (e) {
          return callback(e);
        }
        if (item.type === 'endDatabase') {
          state = 'done';
          return callback(null);
        }
        if (item.type !== 'collection') {
          return callback(new Error('collection expected'));
        }
        var name = item.name;
        var indexes = item.indexes;
        async.series({
          create: function(callback) {
            return db.collection(name, function(err, _collection) {
              if (err) {
                return callback(err);
              }
              collection = _collection;
              return callback(null);
            });
          },
          indexes: function(callback) {
            return async.eachSeries(item.indexes, function(item, callback) {
              if (item.name === '_id_') {
                // automatic
                return setImmediate(callback);
              }
              return collection.ensureIndex(item.key, _.omit(item, 'ns', 'key'), callback);
            }, callback);
          }
        }, function(err) {
          if (err) {
            return callback(err);
          }
          state = 'documents';
          return callback(null);
        });
      },
      documents: function(item, callback) {
        if (version === '1') {
          item = BSON.deserialize(item);
          if (item.type === 'endCollection') {
            state = 'collection';
            return setImmediate(callback);
          }
          if (item.type !== 'document') {
            return callback(new Error('Document expected'));
          }
          return collection.insert(item.document, callback);
        }

        // Just scan for the unique random string that
        // appears in the end-of-collection object. No need
        // to waste time parsing BSON for this. Also
        // leverage the fact that it won't ever be
        // part of a large object to avoid scanning them all
        if (item.length < 100) {
          if (buffertools.indexOf(item, endOfCollection) !== -1) {
            return flush(function(err) {
              if (err) {
                return callback(err);
              }
              state = 'collection';
              return callback(null);
            });
          }
        }
        return async.series({
          flushIfNeeded: function(callback) {
            if (insertQueueSize + item.length <= 16777216) {
              return setImmediate(callback);
            }
            return flush(callback);
          },
          insertInQueue: function(callback) {
            insertQueueSize += item.length;
            insertQueue.push(item);
            return setImmediate(callback);
          }
        }, callback);

        function flush(callback) {
          if (!insertQueue.length) {
            // MongoDB considers it an error to be asked
            // to insert 0 documents ):
            return setImmediate(callback);
          }
          // watch out for race condition
          var insert = insertQueue;
          insertQueue = [];
          insertQueueSize = 0;
          return collection.insert(insert, function(err) {
            insertQueue = [];
            return callback(err);
          });
        }
      }
    };

    return async.series({
      connect: function(callback) {
        if (typeof(dbOrUri) !== 'string') {
          // Already a mongodb connection
          db = dbOrUri;
          return setImmediate(callback);
        }
        return mongodb.MongoClient.connect(dbOrUri, function(err, _db) {
          if (err) {
            return callback(err);
          }
          db = _db;
          return callback(null);
        });
      },
      dropCollections: function(callback) {
        return db.collections(function(err, collections) {
          if (err) {
            return callback(err);
          }
          return async.eachSeries(collections, function(item, callback) {
            if (item.collectionName.match(/^system\./)) {
              return setImmediate(callback);
            }
            return item.drop(callback);
          }, callback);
        });
      },
      loadCollections: function(finalCallback) {
        return reader({ raw: true, from: stream }, function(doc, callback) {
          return processors[state](doc, function(err) {
            if (err) {
              return callback(err);
            }
            if (state === 'done') {
              return finalCallback(null);
            }
            return callback(null);
          });
        }, function(err) {
          if (err) {
            return finalCallback(err);
          }
          return finalCallback(new Error('Premature end of stream'));
        });
      }
    }, callback);
  }
528818
function RequestPQ(callback, channel) {
    flow.runSeries([
        requestPQ,
        findPAndQ,
        findPublicKey,
    ], callback, channel);
}
314208
function getRawBody (stream, options, callback) {
  var done = callback
  var opts = options || {}

  if (options === true || typeof options === 'string') {
    // short cut for encoding
    opts = {
      encoding: options
    }
  }

  if (typeof options === 'function') {
    done = options
    opts = {}
  }

  // validate callback is a function, if provided
  if (done !== undefined && typeof done !== 'function') {
    throw new TypeError('argument callback must be a function')
  }

  // require the callback without promises
  if (!done && !global.Promise) {
    throw new TypeError('argument callback is required')
  }

  // get encoding
  var encoding = opts.encoding !== true
    ? opts.encoding
    : 'utf-8'

  // convert the limit to an integer
  var limit = bytes.parse(opts.limit)

  // convert the expected length to an integer
  var length = opts.length != null && !isNaN(opts.length)
    ? parseInt(opts.length, 10)
    : null

  if (done) {
    // classic callback style
    return readStream(stream, encoding, length, limit, done)
  }

  return new Promise(function executor (resolve, reject) {
    readStream(stream, encoding, length, limit, function onRead (err, buf) {
      if (err) return reject(err)
      resolve(buf)
    })
  })
}
224421
function() {
        activemq = new PodActiveMQ({
          username: params.queueUsername,
          password: params.queuePassword,
          host: params.queueHost,
          port: params.queuePort,
          timeout: params.queueConnectionTimeout
        });

        activemq.on("init", function() {
          fireEvent("asyncReady");

          socketState = socketStateType.OPEN;
          fireEvent("stateChange", {
            queueState: socketState
          });

          pushSendDataQueueHandler();

          activemq.subscribe({
            destination: params.queueReceive,
            ack: "client-individual"
          }, function(message) {
            handleSocketMessage(JSON.parse(message));
            if (onReceiveLogging) {
              asyncLogger("Receive", JSON.parse(message));
            }
          });

        });

        activemq.on("error", function(msg) {
          fireEvent("error", msg);

          socketState = socketStateType.CLOSED;

          fireEvent("stateChange", {
            queueState: socketState
          });
        })
      }
575417
function(options) {
  stream.Duplex.call(this, { objectMode: true });

  this.options = extend(true, Pipeline.DEFAULTS, options || {});
  this.readyState = 0; // 0: closed, 1: open

  this._pipeline = [];

  this.setMaxListeners(0); // unlimited listeners allowed

  this.options.ipc.subscribe(
    this.options.ipc.namespace,
    this._proxy.bind(this)
  );
}
266566
function AmqpTasksSource(amqpOptions) {
  TasksSource.apply(this, arguments);
  assert.ok(amqpOptions, "amqpOptions are required in order to create a AmqpTasksSource");
  Object.defineProperty(this, 'amqpOptions', {
    value: amqpOptions,
    enumerable: false,
    configurable: false
  });
  this.connection = null;
  this._amqpExchange = null;
  this.exchangeName = 'orch';
  this._queues = {};
}
297574
function stringify(opts, callback) {
    assert.object(opts, 'opts');
    assert.func(callback, 'callback');

    let stringified = '';
    const stringifyStream = createStringifyStream(opts);
    const passthroughStream = new stream.PassThrough();
    const cb = once(callback);

    // setup the passthrough stream as a sink
    passthroughStream.on('data', function(chunk) {
        stringified += chunk;
    });

    passthroughStream.on('end', function() {
        return cb(null, stringified);
    });

    // don't know what errors stringify stream may emit, but pass them back
    // up.
    stringifyStream.on('error', function(err) {
        return cb(err);
    });

    stringifyStream.pipe(passthroughStream);
}
596763
function Stream( arr, options ) {
		if ( !Array.isArray( arr ) ) {
			throw new TypeError( 'Stream()::invalid input argument. First argument must be an array.' );
		}
		if ( arguments.length < 2 ) {
			options = {};
		}
		if ( !( this instanceof Stream ) ) {
			return new Stream( arr, options );
		}
		var err = validate( options );
		if ( err ) {
			throw err;
		}
		setOptions( options );
		Readable.call( this, options );
		this._array = arr.slice();
		this._destroyed = false;
		this._enc = options.encoding;
		this._mode = options.objectMode;
		return this;
	}
69399
AmqpClient closeOkConnection(Continuation callback, ErrorHandler error) {
        Object[] args = {};
        AmqpBuffer bodyArg = null;

        String methodName = "closeOkConnection";
        String methodId = "10" + "50";
        AmqpMethod amqpMethod = MethodLookup.LookupMethod(methodId);
        Object[] arguments = {this, amqpMethod, this.id, args, bodyArg};
        asyncClient.enqueueAction(methodName, "write", arguments, callback, error);
        return this;
    }
1268757
function readStream(stream) {
  return new Promise(function(resolve) {
    stream.pipe(concat(function(data) {
      resolve(data.toString().trim());
    }));
  });
}
544694
public void stream(OutputStream os,
                     HeadersAmp headers,
                     String from, 
                     long qId,
                     String address, 
                     String methodName,
                     PodRef podCaller,
                     ResultStream<?> result,
                     Object []args)
    throws IOException
  {
    init(os);

    OutH3 out = _out;

    if (out == null) {
      return;
    }
    
    if (log.isLoggable(Level.FINEST)) {
      log.finest("hamp-stream-w " + methodName + (args != null ? Arrays.asList(args) : "[]")
                + " {to:" + address + ", from:" + from + "}");
    }
    
    out.writeLong(MessageTypeHamp.STREAM.ordinal());
    writeHeaders(out, headers);

    writeFromAddress(out, from);
    out.writeLong(qId);
    
    writeMethod(out, address, methodName, podCaller);

    out.writeObject(result);

    writeArgs(out, args);
    
    //out.flushBuffer();
    out.flush();
  }
715464
function IOStream(options) {
  if (!(this instanceof IOStream)) {
    return new IOStream(options);
  }

  IOStream.super_.call(this, options);

  this.options = options;
  this.id = uuid();
  this.socket = null;

  // Buffers
  this.pushBuffer = [];
  this.writeBuffer = [];

  // Op states
  this._readable = false;
  this._writable = false;
  this.destroyed = false;

  // default to *not* allowing half open sockets
  this.allowHalfOpen = options && options.allowHalfOpen || false;

  this.on('finish', this._onfinish);
  this.on('end', this._onend);
  this.on('error', this._onerror);
}
1324840
function streamToString(callback) {
  var contents = '';

  return through(
    function (chunk, enc, next) {
      contents += chunk.toString('utf8');
      next();
    },
    function (next) { // flush function
      callback.bind(this)(contents, next);
    }
  );
}
793511
function buildStream (host, options, secure) {
  var stream = [];

  // accept options as a secure boolean
  if (options === true) {
    secure = true;
  }

  options = _.defaults(_.isObject(options) ? options : {}, {
    request  : {},
    response : {}
  });

  function applyStreamFunctions (funcSet) {
    _.each(funcSet || [], function (middleware) {
      if (_.isFunction(middleware)) {
        // wrapper allows us to call the callback without passing proxyObj
        stream.push(function asyncWrapper (proxyObj, callback) {
          return middleware(proxyObj, (error, _proxyObj) => callback(error, _proxyObj || proxyObj));
        });
      }
    });
  }

  // if there is `pre` middleware
  let preSet = options.pre;
  if (preSet) {
    // accept raw functions, but wrap it in an array
    preSet = _.isFunction(preSet) ? [preSet] : preSet;
    applyStreamFunctions(preSet);
  }

  // push the core request
  stream.push(proxyReq);

  // if there is `post` middleware
  let postSet = options.post;
  if (postSet) {
    // accept raw functions, but wrap it in an array
    postSet = _.isFunction(postSet) ? [postSet] : postSet;
    applyStreamFunctions(postSet);
  }

  if (secure) {
    options.request.forceHttps = true;
  }

  return stream;
}