# Alternate form of BufferedInputStream

I was having a problem with the TCP/IP socket communications in a web app where the server that the app was talking to would occasionally send bursts of data that would overflow the low level stream buffer, resulting in lost data.

The solution I came up with is basically an unthreaded form of the StreamGobbler and I have been calling it a GreedyBufferedInputStream. The basic idea is that just about any call to this class will include draining the source InputStream.

I'm still fairly new to Scala, so please let me know how this code could be improved. Also, is there anything in the code which could be done more efficiently? Performance is critically important.

package edu.stsci.util

import org.slf4j.Logger

import java.io.InputStream
import java.util

class GreedyBufferedInputStream extends InputStream {
private var logger: Logger = null
private var source: InputStream = null
private var currentBlock: DataBlock = null

def this(logger: Logger, source: InputStream) {
this()

this.logger = logger
this.source = source

drainSource()
}

def this(source: InputStream) {
this(null, source)
}

if (currentBlock == null) -1
}

override def read(destination: Array[Byte], offset: Int, length: Int): Int = {
if (currentBlock == null) return -1

if (currentBlock == null) {// EOF
}
else {
}
}

}

override def skip(length: Long): Long = {
if (currentBlock == null) return -1

var bytesSkipped = currentBlock.skip(length)

while (bytesSkipped < length) {
if (currentBlock == null) { // EOF
return bytesSkipped
}
else {
bytesSkipped += currentBlock.skip(length - bytesSkipped)
}
}

bytesSkipped
}

override def close() {
super.close()
}

override def available() = {
drainSource()

var result = 0

if (currentBlock != null) result += currentBlock.available

val it = data.iterator()
while (it.hasNext) {
val next = it.next()
result += next.available
}

result
}

private def drainSource() {
if (source == null) return // EOF

if (source.available() > 0) {
val raw = new Array[Byte](source.available())
val block = new DataBlock(raw, length)
}
}

drainSource()

if (currentBlock != null) {
val done = currentBlock.isDone
if (done) currentBlock = null
else return // we have a current block
}

if (data.isEmpty) { // no choice but to block
if (source == null) return // have reached EOF

val raw = new Array[Byte](1024)
if (length < 0) {
source = null
return
}
currentBlock = new DataBlock(raw, length)

}
else currentBlock = data.remove()
}
}

class DataBlock(data: Array[Byte], length: Int) {

def isDone: Boolean = (readPos >= length)

def available: Int = {
}

raw & 0xff
}

def read(offset: Int, length: Int, destination: Array[Byte]): Int = {

if (available >= length) length
else available
}

}

def skip(length: Long): Long = {

if (available >= length) length
else available
}

}
}

• Would it be at all helpful if I added the unit tests? Jul 17, 2012 at 14:30

I think that the use of a LinkedList is not optimal. In every invocation of drainSource you potentially add a new DataBlock ant the end of the linked list; this has performance O(n) (length of the list). I propose to use a Vector instead of the LinkedList.

Another point (but this is not Scala specific): Why do you initialize the field readPos in class DataBlock with -1? You have many special cases due to this choice. I would implement class DataBlock as follows:

class DataBlock(data: Array[Byte], length: Int) {
require(length > 0)

def isDone: Boolean = readPos >= length

def available: Int = length - readPos

def read(offset: Int, length: Int, destination: Array[Byte]): Int = {
val readCount = length min available
}

def skip(length: Long): Long = {
val readCount = length min available
}
}

• I used java.util.LinkedList as I believe that it offers constant time adds at the end as well as constant time removes from the beginning. I dropped your DataBlock implementation in and it passed all the tests. Jul 18, 2012 at 17:53
• @Donald.McLean you are right, missed that you use a Java list and not the Scala one. Jul 18, 2012 at 19:48

Just some random bits of feedback, not really addressing the code as a whole.

First, I would handle constructing the class as follows:

class GreedyBufferedInputStream(logger: Logger = null, initialSource: InputStream)
extends InputStream {

private var source: InputStream = initialSource
private var currentBlock: DataBlock = null

drainSource()


This eliminates a var and both auxiliary constructors.

I would rewrite the available() method as follows:

  override def available() = {
drainSource()
import collection.JavaConverters._
data.asScala.map(_.available).sum +
Option(currentBlock).map(_.available).getOrElse(0)
}


And I would replace this:

var readCount = {
if (available >= length) length
else available
}


with

val readCount = length min available

• I like this answer, but I thought that Dominik's answer was slightly more useful. If I could, I would have split the bonus. Jul 20, 2012 at 11:53

To be precise I have no real solution to your problem, but if I think about it there should be a few possibilities to find one: First you could think about limiting data send by the App. A limitation could mean a two way communication and/or a sending buffer on side of the App. So you can regulate the data sent and maybe zip it to be more efficient etc. A second solution could be a pool of multiple stream buffers which can be chosen in case of a data burst to handle all incomming data. Because I am a Python programmer I cant tell you if your Scala code can be improved or not. I hope I could help a bit.

• Thank you for your answer. However, I tried to be clear - the code I gave is the answer to my original problem and I am now looking for a review of this code. Jul 16, 2012 at 11:54

No time to read it in any detail, but as a general observation of scala style most scala users would be looking to make far more use of immutable ways of doing things. I use var very sparingly these days and never use null as an indicator of failure or empty something - I use Option[X] instead.

• I actually understand Option - it's a startlingly useful construct. However, I fail to see, within the context of this class, where using Option would add any value. I could be wrong, but you'll have to provide some actual justification to convince me. Jul 18, 2012 at 14:44
• I specifically said I have no time to look in detail at your classes. You may be right. Null creeps in from java - I try to wrap it up as soon as possible. Jul 18, 2012 at 15:01