Improve OFStreamObserver.

It also looks at the cache of each stream now and does not block if
there is still data in the cache.