Consume data until you get a new line, emit a line event, clean buffer, rinse, repeat
327986
function StreamHandler(host, port, delimiter) {
	events.EventEmitter.call(this);

	var self = this;
	self.delimiter = delimiter || "\r\n";
	self.host      = host;
	self.port      = port;

	self.buffer    = '';

	self.conn = net.createConnection(self.port, self.host);
	self.conn.setEncoding("utf8");

	self.conn.on('error', function(err) {
		try {
			self.emit('error', 'Stream error: ' + err, err);
		} catch (e) {
			console.log('' + err);
		}
	});
	self.conn.on('data', function(data) {
		self.conn.pause();
		self.buffer += data;
		if (self.buffer.match(/\n/)) {
			var arr = self.buffer.split(self.delimiter);
			//Everything except the last one
			for(var i = 0; i < arr.length - 1; i++) {
				self.emit('line', arr[i]);
			}
			self.buffer = arr[arr.length - 1];
		}
		if (self.conn.readyState == 'open') {
			self.conn.resume();
		}
	});
	self.conn.on('close', function(had_error) {
		if (!had_error && self.buffer.length > 0) {
			self.emit('line', self.buffer);
		}
	});
}
975009
def pipe_data_received(self, fd, data):
        """"""
        if fd == 2:  # stderr fd number
            self._on_stderr(data)
        elif self._on_data:
            self._on_data(data)
        else:
            self._queued_data.append(data)
1367095
function( data ) {
   
            
            var read_buffer = read_buffers_by_connection[tcp_connection];
            read_buffer += data;

            //split by new lines, 
            var messages = read_buffer.split('\n');
            
            //last element is either a partial message, or ''            
            read_buffer = messages.pop();
        
            for ( var i = 0; i < messages.length; i++ ) {
                
                //TODO error handling
                var message = messages[i];
                var components = message.split( ' ' );
                var path = components[0];
                var value = components[1];
                var timestamp = components[2];
                
                that.emit(  path, value, timestamp );                
            }            
        }
1278115
function LineInfoEmitter(options, rules) {
    var stdin = process.stdin,
        incompleteLine,
        self = this;

    EventEmitter.call(this);

    // prepare stdin for character processing and listen to events
    stdin.setEncoding(options.encoding);

    function processData(data) {
        var lastLineIncomplete, lines;

        if (options.debug) {
            console.log('====> incompleteLine:', incompleteLine, 'new chunk:');
            hexdumpLine(data);
        }

        if (incompleteLine) {
            // append data to not yet emitted, incomplete line
            data = incompleteLine + data;
            incompleteLine = null;
        }

        lastLineIncomplete = data && data.charAt(data.length - 1) !== '\n';
        lines = data.split('\n');

        if (lastLineIncomplete) {
            // wait to emit last line until rest of line comes in
            incompleteLine = lines.pop();
        }
        else {
            // remove last '' after newline
            lines.pop();
        }
        lines.forEach(function (line, i) {
            if (options.debug) {
                hexdumpLine(line);
            }
            var lineInfo = analyzeLine(line, rules);
            self.emit('lineInfo', lineInfo);
        });
    }

    stdin.on('data', processData);

    stdin.on('end', function () {
        if (incompleteLine) {
            processData('\n');
        }
        self.emit('end');
    });

    // stdin is paused by default, so let's start!
    stdin.resume();
}
1223009
def _flush_raw_or_buffered(self):
        
        # Flush only if bytes written
        # This avoid no required process/thread
        # creation and network call.
        # This step is performed by raw stream.
        if self._buffer_seek and self._seek:
            self._seek += 1
            with handle_os_exceptions():
                self._flush()

        # If data lower than buffer size
        # flush data with raw stream to reduce IO calls
        elif self._buffer_seek:
            self._raw._write_buffer = self._get_buffer()
            self._raw._seek = self._buffer_seek
            self._raw.flush()
322295
def process_data(self):
        ""

        try:
            reader = getattr(self.socket, 'read', self.socket.recv)
            new_data = reader(2 ** 14)
        except socket.error:
            # The server hung up.
            self.disconnect("Connection reset by peer")
            return
        if not new_data:
            # Read nothing: connection must be down.
            self.disconnect("Connection reset by peer")
            return

        self.buffer.feed(new_data)

        # process each non-empty line after logging all lines
        for line in self.buffer:
            log.debug("FROM SERVER: %s", line)
            if not line:
                continue
            self._process_line(line)
477682
async def on_data(self, data):
        
        self._receive_buffer += data

        # Schedule new timeout event.
        if self._ping_checker_handle:
            self._ping_checker_handle.cancel()

        # create a task for the ping checker
        self._ping_checker_handle = self.eventloop.create_task(
            self._perform_ping_timeout(self.PING_TIMEOUT))

        while self._has_message():
            message = self._parse_message()
            self.eventloop.create_task(self.on_raw(message))
267550
def lines_from_stream(chunks):
    
    buf = buffer.DecodingLineBuffer()
    for chunk in chunks:
        buf.feed(chunk)
        # when Python 3, yield from buf
        for _ in buf:
            yield _
970450
function processByte (stream, b) {
    assert.equal(typeof b, 'number');
    if (b === NEWLINE) {
        stream.emit('newline');
    }
}
119772
def feed(self, data):
        
        self._lock.acquire()
        try:
            if self._event is not None:
                self._event.set()
            self._buffer_frombytes(b(data))
            self._cv.notifyAll()
        finally:
            self._lock.release()
1061494
function LineReader(file) {
  if (typeof file === 'string') {
    this.readstream = fs.createReadStream(file);
  } else {
    this.readstream = file;
  }
  this.remainBuffers = [];
  var self = this;
  this.readstream.on('data', function (data) {
    self.ondata(data);
  });
  this.readstream.on('error', function (err) {
    self.emit('error', err);
  });
  this.readstream.on('end', function () {
    self.emit('end');
  });
}
888540
def frameify(self, state, data):
        """"""

        # Pull in any partially-processed data
        data = state.recv_buf + data

        # Loop over the data
        while data:
            line, sep, rest = data.partition('\n')

            # Did we have a whole line?
            if sep != '\n':
                break

            # OK, update the data...
            data = rest

            # Now, strip off carriage return, if there is one
            if self.carriage_return and line[-1] == '\r':
                line = line[:-1]

            # Yield the line
            try:
                yield line
            except FrameSwitch:
                break

        # Put any remaining data back into the buffer
        state.recv_buf = data
436459
void afterRead(Node<K, V> node, long now, boolean recordHit) {
    if (recordHit) {
      statsCounter().recordHits(1);
    }

    boolean delayable = skipReadBuffer() || (readBuffer.offer(node) != Buffer.FULL);
    if (shouldDrainBuffers(delayable)) {
      scheduleDrainBuffers();
    }
    refreshIfNeeded(node, now);
  }
1033455
function createBufferStream (stream, interval) {
  var buf = []
  var timer = null

  // flush function
  function flush () {
    timer = null
    stream.write(buf.join(''))
    buf.length = 0
  }

  // write function
  function write (str) {
    if (timer === null) {
      timer = setTimeout(flush, interval)
    }

    buf.push(str)
  }

  // return a minimal "stream"
  return { write: write }
}
450841
def get_raw_data_from_buffer(self, filter_func=None, converter_func=None):
        '''
        '''
        if self._is_running:
            raise RuntimeError('Readout thread running')
        if not self.fill_buffer:
            logging.warning('Data buffer is not activated')
        return [convert_data_array(data_array_from_data_iterable(data_iterable), filter_func=filter_func, converter_func=converter_func) for data_iterable in self._data_buffer]
356454
def push_line(self, line):
        
        self.lineno -= 1
        self._buffer.append(line + b'\n')
795939
def write_to_buffer
      emit_comments_before if buffer.fresh_line?
      dispatch
      comments.consume(node)
      emit_eof_comments if parent.is_a?(Root)
      self
    end