* Copyright (c) Huawei Technologies Co., Ltd. 2025. All rights reserved.
* You can use this software according to the terms and conditions of the Mulan PSL v2.
* You may obtain a copy of Mulan PSL v2 at:
* http://license.coscl.org.cn/MulanPSL2
* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
* EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
* MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
* See the Mulan PSL v2 for more details.
*/
#pragma once
#include <securec.h>
#include <vector>
#include <cstddef>
#include <cstdint>
#include <cstring>
#include <utils/ByteBuffer.h>
#include "include/outputbuffer.h"
#include "include/common.h"
#include "basictypes/Double.h"
class DataOutputSerializer {
public:
DataOutputSerializer() = default;
explicit DataOutputSerializer(int size);
~DataOutputSerializer()
{
if (data_ != nullptr && (this->outputBufferStatus == nullptr || this->outputBufferStatus->ownership == 1)) {
delete[] data_;
}
}
inline void setBackendBuffer(uint8_t* address, int32_t capacity);
inline void setBackendBuffer(OutputBufferStatus* outputBufferStatus);
inline void clear();
inline void setPosition(int position);
inline void setPositionUnsafe(int position);
inline void writeIntUnsafe(uint32_t value, int pos);
inline int length() const;
inline void writeLong(int64_t value);
inline void writeRecordTimestamp(uint64_t value);
inline void writeByte(uint32_t var);
inline void writeUTF(std::string str);
inline void write(uint32_t var1);
inline void writeShort(uint16_t value);
inline void write(std::vector<uint8_t>& var);
inline void writeInt(uint32_t value);
inline void writeBoolean(bool var);
inline void writeDouble(double value);
inline void write(uint8_t* buffer, int bufferSize, int dstOffset, int length);
inline void expandDataBuffer(int requiredSize);
inline ByteBuffer* wrapAsByteBuffer();
inline uint8_t* getData();
inline int getPosition();
std::vector<uint8_t>* getCopyOfBuffer()
{
return new std::vector<uint8_t>(data_, data_ + position_);
}
private:
uint8_t* data_ = nullptr;
int position_ = 0;
int capacity_ = 0;
OutputBufferStatus* outputBufferStatus = nullptr;
ByteBuffer reusedWrapper;
};
inline DataOutputSerializer::DataOutputSerializer(int size)
{
this->data_ = new uint8_t[size];
}
inline ByteBuffer* DataOutputSerializer::wrapAsByteBuffer()
{
ByteBuffer* byteBuffer = reusedWrapper.wrapOpt(data_, 0, position_);
return byteBuffer;
}
inline uint8_t* DataOutputSerializer::getData()
{
return data_;
}
inline void DataOutputSerializer::setBackendBuffer(uint8_t* address, int32_t capacity)
{
data_ = address;
capacity_ = capacity;
position_ = 0;
}
inline void DataOutputSerializer::setBackendBuffer(OutputBufferStatus* outputBufferStatus_)
{
data_ = reinterpret_cast<uint8_t*>(outputBufferStatus_->outputBuffer_);
capacity_ = outputBufferStatus_->capacity_;
position_ = 0;
this->outputBufferStatus = outputBufferStatus_;
}
inline void DataOutputSerializer::clear()
{
position_ = 0;
}
inline void DataOutputSerializer::setPosition(int position)
{
position_ = position;
}
inline void DataOutputSerializer::setPositionUnsafe(int position)
{
position_ = position;
}
inline void DataOutputSerializer::writeIntUnsafe(uint32_t value, int pos)
{
*reinterpret_cast<uint32_t*>(data_ + pos) = __builtin_bswap32(value);
}
inline int DataOutputSerializer::length() const
{
return position_;
}
inline void DataOutputSerializer::writeByte(uint32_t var)
{
expandDataBuffer(1);
*(data_ + position_++) = var & 0xff;
}
inline void DataOutputSerializer::write(uint32_t value)
{
writeByte(value);
}
inline void DataOutputSerializer::writeShort(uint16_t value)
{
expandDataBuffer(2);
if (data_ == nullptr) {
throw std::runtime_error("Data buffer is null");
}
*reinterpret_cast<uint16_t*>(data_ + position_) = __builtin_bswap16(value);
position_ += 2;
}
inline void DataOutputSerializer::write(std::vector<uint8_t>& var)
{
write(var.data(), var.size(), 0, var.size());
}
inline void DataOutputSerializer::writeInt(uint32_t value)
{
expandDataBuffer(4);
if (data_ == nullptr) {
throw std::runtime_error("Data buffer is null");
}
*reinterpret_cast<uint32_t*>(data_ + position_) = __builtin_bswap32(value);
position_ += 4;
}
inline void DataOutputSerializer::writeLong(int64_t value)
{
expandDataBuffer(8);
if (data_ == nullptr) {
throw std::runtime_error("Data buffer is null");
}
*reinterpret_cast<int64_t*>(data_ + position_) = __builtin_bswap64(value);
position_ += 8;
}
* Write timestamp in a record element
* The timestamp is written in big endian format, while the rest of the record element is written in little endian
* format Thus, we need this specific function
*/
inline void DataOutputSerializer::writeRecordTimestamp(uint64_t value)
{
expandDataBuffer(8);
if (data_ == nullptr) {
throw std::runtime_error("Data buffer is null");
}
*reinterpret_cast<uint64_t*>(data_ + position_) = __builtin_bswap64(value);
position_ += 8;
}
inline void DataOutputSerializer::write(uint8_t* buffer, int bufferSize, int dstOffset, int length)
{
expandDataBuffer(length);
std::copy(buffer + dstOffset, buffer + dstOffset + length, data_ + position_);
position_ += length;
}
inline void DataOutputSerializer::expandDataBuffer(int requiredSize)
{
if (unlikely(position_ + requiredSize > capacity_)) {
LOG("******************output buffer is full, expand the buffer****************************************");
capacity_ = (position_ + requiredSize) * 4;
uint8_t* newData = new uint8_t[capacity_];
if (likely(data_ != nullptr)) {
std::copy(data_, data_ + position_, newData);
}
if (outputBufferStatus != nullptr && outputBufferStatus->ownership == 0) {
STD_LOG("data_ is owned by java, should not delete " << reinterpret_cast<uintptr_t>(data_));
} else {
delete[] data_;
}
data_ = newData;
if (outputBufferStatus != nullptr) {
this->outputBufferStatus->ownership = 1;
this->outputBufferStatus->outputBuffer_ = reinterpret_cast<uintptr_t>(data_);
this->outputBufferStatus->capacity_ = capacity_;
}
}
}
inline int DataOutputSerializer::getPosition()
{
return position_;
}
inline void DataOutputSerializer::writeUTF(std::string str)
{
int strlen = str.length();
int utflen = 0;
int c;
char* strChar = str.data();
for (int i = 0; i < strlen; i++) {
c = strChar[i];
if ((c >= 0x0001) && (c <= 0x007F)) {
utflen++;
} else if (c > 0x07FF) {
utflen += 3;
} else {
utflen += 2;
}
}
if (utflen > 65535) {
THROW_RUNTIME_ERROR("Encoded string is too long: " + utflen);
}
expandDataBuffer(utflen + 2);
int count = position_;
*(data_ + count++) = (utflen >> 8) & 0xff;
*(data_ + count++) = utflen & 0xff;
int i;
for (i = 0; i < strlen; i++) {
c = strChar[i];
if (!((c >= 0x0001) && (c <= 0x007F))) {
break;
}
*(data_ + count++) = c;
}
for (; i < strlen; i++) {
c = strChar[i];
if ((c >= 0x0001) && (c <= 0x007F)) {
*(data_ + count++) = c;
} else if (c > 0x07FF) {
*(data_ + count++) = (0xE0 | ((c >> 12) & 0x0F));
*(data_ + count++) = (0x80 | ((c >> 6) & 0x3F));
*(data_ + count++) = (0x80 | (c & 0x3F));
} else {
*(data_ + count++) = (0xC0 | ((c >> 6) & 0x1F));
*(data_ + count++) = (0x80 | (c & 0x3F));
}
}
position_ = count;
}
inline void DataOutputSerializer::writeDouble(double value)
{
writeLong(Double::doubleToLongBits(value));
}
inline void DataOutputSerializer::writeBoolean(bool var)
{
write(var ? 1 : 0);
}