CLOUD: Rewrite NetworkReadStream

Now it is based on MemoryReadWriteStream, which is introduced by this
commit. This stream is using ring buffer and is dynamically increasing
its size when necessary.

1 files changed, 83 insertions, 0 deletions

diff --git a/common/memstream.h b/common/memstream.h
index 59d5f15b1a..963637cb27 100644
--- a/common/memstream.h
+++ b/common/memstream.h
@@ -209,6 +209,89 @@ public:
 	bool seek(int32 offset, int whence = SEEK_SET);
 };
 
+/**
+* MemoryStream based on RingBuffer. Grows if has insufficient buffer size.
+*/
+class MemoryReadWriteStream : public WriteStream {
+private:
+	uint32 _capacity;
+	uint32 _size;
+	byte *_data;
+	uint32 _writePos, _readPos, _pos, _length;
+	DisposeAfterUse::Flag _disposeMemory;
+
+	void ensureCapacity(uint32 new_len) {
+		if (new_len <= _capacity)
+			return;
+
+		byte *old_data = _data;
+		uint32 oldCapacity = _capacity;
+
+		_capacity = MAX(new_len + 32, _capacity * 2);
+		_data = (byte *)malloc(_capacity);
+
+		if (old_data) {			
+			// Copy old data
+			if (_readPos < _writePos) {
+				memcpy(_data, old_data + _readPos, _writePos - _readPos);
+				_writePos -= _readPos;
+				_readPos = 0;
+			} else {
+				memcpy(_data, old_data + _readPos, oldCapacity - _readPos);
+				memcpy(_data + oldCapacity - _readPos, old_data, _writePos);
+				_writePos += oldCapacity - _readPos;
+				_readPos = 0;
+			}
+			free(old_data);
+		}
+	}
+public:
+	MemoryReadWriteStream(DisposeAfterUse::Flag disposeMemory = DisposeAfterUse::NO) : _capacity(0), _size(0), _data(0), _writePos(0), _readPos(0), _pos(0), _length(0), _disposeMemory(disposeMemory) {}
+
+	~MemoryReadWriteStream() {
+		if (_disposeMemory)
+			free(_data);
+	}
+
+	uint32 write(const void *dataPtr, uint32 dataSize) {
+		ensureCapacity(_length + dataSize);
+		if (_writePos + dataSize < _capacity) {
+			memcpy(_data + _writePos, dataPtr, dataSize);
+		} else {
+			memcpy(_data + _writePos, dataPtr, _capacity - _writePos);
+			const byte *shiftedPtr = (const byte *)dataPtr + _capacity - _writePos;
+			memcpy(_data, shiftedPtr, dataSize - (_capacity - _writePos));
+		}
+		_writePos = (_writePos + dataSize) % _capacity;
+		_pos += dataSize;
+		_length += dataSize;
+		if (_pos > _size)
+			_size = _pos;
+		return dataSize;
+	}
+
+	uint32 read(void *dataPtr, uint32 dataSize) {
+		uint32 length = _length;
+		if (length < dataSize) dataSize = length;
+		if (dataSize == 0 || _capacity == 0) return 0;
+		if (_readPos + dataSize < _capacity) {
+			memcpy(dataPtr, _data + _readPos, dataSize);
+		} else {
+			memcpy(dataPtr, _data + _readPos, _capacity - _readPos);
+			byte *shiftedPtr = (byte *)dataPtr + _capacity - _readPos;
+			memcpy(shiftedPtr, _data, dataSize - (_capacity - _readPos));
+		}
+		_readPos = (_readPos + dataSize) % _capacity;
+		_length -= dataSize;
+		return dataSize;
+	}
+
+	uint32 pos() const { return _pos - _length; } //'read' position in the stream
+	uint32 size() const { return _size; } //that's also 'write' position in the stream, as it's append-only
+
+	byte *getData() { return _data; }	
+};
+
 } // End of namespace Common
 
 #endif