Override this to handle the streams as they arrive

        :param IOStream stream: An IOStream for processing

        See https://tornado.readthedocs.io/en/latest/iostream.html#tornado.iostream.IOStream
        for additional details.
485621
def handle_stream(self, stream):
        '''
        
        '''
        @tornado.gen.coroutine
        def _null(msg):
            raise tornado.gen.Return(None)

        def write_callback(stream, header):
            if header.get('mid'):
                @tornado.gen.coroutine
                def return_message(msg):
                    pack = salt.transport.frame.frame_msg_ipc(
                        msg,
                        header={'mid': header['mid']},
                        raw_body=True,
                    )
                    yield stream.write(pack)
                return return_message
            else:
                return _null
        if six.PY2:
            encoding = None
        else:
            encoding = 'utf-8'
        unpacker = msgpack.Unpacker(encoding=encoding)
        while not stream.closed():
            try:
                wire_bytes = yield stream.read_bytes(4096, partial=True)
                unpacker.feed(wire_bytes)
                for framed_msg in unpacker:
                    body = framed_msg['body']
                    self.io_loop.spawn_callback(self.payload_handler, body, write_callback(stream, framed_msg['head']))
            except tornado.iostream.StreamClosedError:
                log.trace('Client disconnected from IPC %s', self.socket_path)
                break
            except socket.error as exc:
                # On occasion an exception will occur with
                # an error code of 0, it's a spurious exception.
                if exc.errno == 0:
                    log.trace('Exception occured with error number 0, '
                              'spurious exception: %s', exc)
                else:
                    log.error('Exception occurred while '
                              'handling stream: %s', exc)
            except Exception as exc:
                log.error('Exception occurred while '
                          'handling stream: %s', exc)
20637
public void write(PrintStream stream) {
        this.output = stream;
        dispatch(base, false);
        stream.flush();
    }
679224
def handle_stream(self, message):
        
        logging.debug("handle_stream(...)")

        event = Event()
        for line in message.strip().splitlines():
            (field, value) = line.split(":", 1)
            field = field.strip()

            if field == "event":
                event.name = value.lstrip()
            elif field == "data":
                value = value.lstrip()
                if event.data is None:
                    event.data = value
                else:
                    event.data = "%s\n%s" % (event.data, value)
            elif field == "id":
                event.id = value.lstrip()
                self.last_event_id = event.id
            elif field == "retry":
                try:
                    self.retry_timeout = int(value)
                    event.retry = self.retry_timeout
                    logging.info("timeout reset: %s" % (value,))
                except ValueError:
                    pass
            elif field == "":
                logging.debug("received comment: %s" % (value,))
            else:
                raise Exception("Unknown field !")
        self.events.append(event)
1338953
def build_stream(self, obj, stream: BytesIO, context=None) -> None:
        
        if context is None:
            context = Context()
        if not isinstance(context, Context):
            context = Context(context)
        try:
            self._build_stream(obj, stream, context)
        except Error:
            raise
        except Exception as exc:
            raise BuildingError(str(exc))
862864
def _handle_stream_line(line):
    
    # XXX This necessary processing is probably a bug from docker-py,
    # hence, might break if the bug is fixed, i.e. we should get decoded JSON
    # directly from the API.
    line = json.loads(line)
    if 'error' in line:
        content = "ERROR: " + line['error']
    else:
        content = line.get('stream', '')
    for streamline in content.split('\n'):
        if streamline:
            yield streamline
299952
def from_stream(cls, stream, offset):
        
        ifd_parser = _IfdParser(stream, offset)
        entries = dict((e.tag, e.value) for e in ifd_parser.iter_entries())
        return cls(entries)
485134
public void listStream(PRStream stream, PdfReaderInstance reader)
    {
        try {
            listDict(stream);
            out.println("startstream");
            byte[] b = PdfReader.getStreamBytes(stream);
//                  byte buf[] = new byte[Math.min(stream.getLength(), 4096)];
//                  int r = 0;
//                  stream.openStream(reader);
//                  for (;;) {
//                      r = stream.readStream(buf, 0, buf.length);
//                      if (r == 0) break;
//                      out.write(buf, 0, r);
//                  }
//                  stream.closeStream();
            int len = b.length - 1;
            for (int k = 0; k < len; ++k) {
                if (b[k] == '\r' && b[k + 1] != '\n')
                    b[k] = (byte)'\n';
            }
            out.println(new String(b));
            out.println("endstream");
        } catch (IOException e) {
            System.err.println("I/O exception: " + e);
//          } catch (java.util.zip.DataFormatException e) {
//              System.err.println("Data Format Exception: " + e);
        }
    }
1360723
@Override
  public StreamImpl stream()
    throws IOException
  {
    if (_streamImpl == null) {
      _streamImpl = new SocketStream();
    }

    _streamImpl.init(_s);

    return _streamImpl;
  }
105520
def stream=(stream)
      logger = Logger.new(stream)
      logger.level = @logger.level
      logger.formatter = @logger.formatter
      @logger = logger
    end
499732
def parse(stream, Loader=Loader):
    
    loader = Loader(stream)
    try:
        while loader.check_event():
            yield loader.get_event()
    finally:
        loader.dispose()
1070920
def stream_request(self, request, out_future):
        """"""
        request.close_argstreams()

        def on_done(future):
            if future.exception() and out_future.running():
                out_future.set_exc_info(future.exc_info())
            request.close_argstreams(force=True)

        stream_future = self._stream(request, self.request_message_factory)
        stream_future.add_done_callback(on_done)
        return stream_future
1213393
def start_event_stream(self):
        
        if not self._stream:
            self._stream = GerritStream(self, ssh_client=self._ssh_client)
            self._stream.start()
46080
private void streamToOutput(InputStream inputStream) throws IOException {
    try {
      byte[] buffer = new byte[4096];
      int bytesRead = -1;
      while ((bytesRead = inputStream.read(buffer)) != -1) {
        outputStream.write(buffer, 0, bytesRead);
      }
      outputStream.flush();
    } finally {
      inputStream.close();
    }
  }
30243
private void startTwitterStream() {
        logger.info("starting {} twitter stream", streamType);

        if (stream == null) {
            logger.debug("creating twitter stream");

            stream = new TwitterStreamFactory(buildTwitterConfiguration()).getInstance();
            if (streamType.equals("user")) {
                stream.addListener(new UserStreamHandler());
            } else {
                stream.addListener(new StatusHandler());
            }

            logger.debug("twitter stream created");
        }

        if (riverStatus != RiverStatus.STOPPED && riverStatus != RiverStatus.STOPPING) {
            if (streamType.equals("filter") || filterQuery != null) {
                stream.filter(filterQuery);
            } else if (streamType.equals("firehose")) {
                stream.firehose(0);
            } else if (streamType.equals("user")) {
                stream.user();
            } else {
                stream.sample();
            }
        }
        logger.debug("{} twitter stream started!", streamType);
    }
1140451
function ChunkStreamWithStream(stream, options) {
  ChunkStream.call(this, options);

  stream.pause(); // Pause stream and wait for data listener. It's useful for node v0.8
  this._stream = stream;
  this._stream.on('end', this.end.bind(this)); // Should catch the end event for node v0.8
}
824263
def alert_stream(self, reset_event, kill_event):
        """"""
        _LOGGING.debug('Stream Thread Started: %s, %s', self.name, self.cam_id)
        start_event = False
        parse_string = ""
        fail_count = 0

        url = '%s/ISAPI/Event/notification/alertStream' % self.root_url

        # pylint: disable=too-many-nested-blocks
        while True:

            try:
                stream = self.hik_request.get(url, stream=True,
                                              timeout=(CONNECT_TIMEOUT,
                                                       READ_TIMEOUT))
                if stream.status_code == requests.codes.not_found:
                    # Try alternate URL for stream
                    url = '%s/Event/notification/alertStream' % self.root_url
                    stream = self.hik_request.get(url, stream=True)

                if stream.status_code != requests.codes.ok:
                    raise ValueError('Connection unsucessful.')
                else:
                    _LOGGING.debug('%s Connection Successful.', self.name)
                    fail_count = 0
                    self.watchdog.start()

                for line in stream.iter_lines():
                    # _LOGGING.debug('Processing line from %s', self.name)
                    # filter out keep-alive new lines
                    if line:
                        str_line = line.decode("utf-8", "ignore")
                        # New events start with --boundry
                        if str_line.find('<EventNotificationAlert') != -1:
                            # Start of event message
                            start_event = True
                            parse_string += str_line
                        elif str_line.find('</EventNotificationAlert>') != -1:
                            # Message end found found
                            parse_string += str_line
                            start_event = False
                            if parse_string:
                                tree = ET.fromstring(parse_string)
                                self.process_stream(tree)
                                self.update_stale()
                                parse_string = ""
                        else:
                            if start_event:
                                parse_string += str_line

                    if kill_event.is_set():
                        # We were asked to stop the thread so lets do so.
                        break
                    elif reset_event.is_set():
                        # We need to reset the connection.
                        raise ValueError('Watchdog failed.')

                if kill_event.is_set():
                    # We were asked to stop the thread so lets do so.
                    _LOGGING.debug('Stopping event stream thread for %s',
                                   self.name)
                    self.watchdog.stop()
                    self.hik_request.close()
                    return
                elif reset_event.is_set():
                    # We need to reset the connection.
                    raise ValueError('Watchdog failed.')

            except (ValueError,
                    requests.exceptions.ConnectionError,
                    requests.exceptions.ChunkedEncodingError) as err:
                fail_count += 1
                reset_event.clear()
                _LOGGING.warning('%s Connection Failed (count=%d). Waiting %ss. Err: %s',
                                 self.name, fail_count, (fail_count * 5) + 5, err)
                parse_string = ""
                self.watchdog.stop()
                self.hik_request.close()
                time.sleep(5)
                self.update_stale()
                time.sleep(fail_count * 5)
                continue
59899
def _begin_stream(self, command: Command):
        ''''''
        begin_reply = yield from self._commander.begin_stream(command)

        self._response.reply = begin_reply

        self.event_dispatcher.notify(self.Event.begin_transfer, self._response)