    virtual ROBuffer getNext() {
        if (!base->vbuffers[stream_id].empty()) {
            auto retval = base->vbuffers[stream_id].front();
            if (!retval->empty()) {
                base->vbuffers[stream_id].pop();
            }
            if (retval->empty()) {
                isEof = true;
                return {NULL, NULL};
            }
            return {retval->data(), retval->data() + retval->size()};
        }
        while(!isEof) {
            auto dat = worker->batch_recv_data();
            for (unsigned int i = 0; i < dat.count; ++i) {
                ResizableByteBufferListNode* lnode = (ResizableByteBufferListNode*) dat.data[i];
                if (dat.count == 1 && lnode->stream_id == stream_id && lnode && lnode->size()) {
                    assert(stream_id == lnode->stream_id);
                    last = lnode;
                    return {lnode->data(), lnode->data() + lnode->size()};
                } else {
                    base->vbuffers[lnode->stream_id].push(lnode);
                }
            }
            if (!base->vbuffers[stream_id].empty()) {
                return getNext(); // recursive call, 1 deep
            }
            if (dat.return_code < 0) {
                isEof = true; // hmm... should we bail here?
                always_assert(false);
            }
        }
        return {NULL, NULL};
    }
    virtual ROBuffer getNext() {
        if (!base->vbuffers[stream_id].empty()) {
            auto retval = base->vbuffers[stream_id].front();
            if (!retval->empty()) {
                base->vbuffers[stream_id].pop();
            }
            if (retval->empty()) {
                isEof = true;
                return {NULL, NULL};
            }
            return {retval->data(), retval->data() + retval->size()};
        }
        while(!isEof) {
            auto dat = worker->batch_recv_data();
            for (unsigned int i = 0; i < dat.count; ++i) {
                ResizableByteBufferListNode* lnode = (ResizableByteBufferListNode*) dat.data[i];
                if (dat.count == 1 && lnode->stream_id == stream_id && lnode && lnode->size()) {
                    assert(stream_id == lnode->stream_id);
                    last = lnode;
                    return {lnode->data(), lnode->data() + lnode->size()};
                } else {
                    base->vbuffers[lnode->stream_id].push(lnode);
                }
            }
            if (!base->vbuffers[stream_id].empty()) {
                return getNext();  
            }
            if (dat.return_code < 0) {
                isEof = true;  
                always_assert(false);
            }
        }
        return {NULL, NULL};
    }