Skip to content

O
oblogmsg

oceanbase / oblogmsg
9 个月 前同步成功

通知 1
Star 10
Fork 6
O
oblogmsg

前往新版Gitcode，体验更适合开发者的 AI 搜索 >>

提交 bf5d80ed 编写于 作者: 忠 忠阳 提交者： xiaoyi.yl
浏览文件
操作

impl oblogmsg

上级 47777a85
展开全部 隐藏空白更改
内联 并排
Showing with 9396 addition and 0 deletion
CMakeLists.txt 0 → 100644
浏览文件 @ bf5d80ed
cmake_minimum_required(VERSION 3.14)
project(oblogmsg)
option(TEST "build tests" OFF)
option(BENCHMARK "build benchmarks" OFF)
add_definitions(-D_GLIBCXX_USE_CXX11_ABI=0)
set(CMAKE_VERBOSE_MAKEFILE OFF)
if (NOT CMAKE_BUILD_TYPE)
set(CMAKE_BUILD_TYPE "RelWithDebInfo")
endif ()
add_library(oblogmsg_base INTERFACE)
set(INNER_INCLUDE_DIRS ${CMAKE_CURRENT_SOURCE_DIR}/include)
set_property(GLOBAL PROPERTY OBLOGMSG_INCLUDE_DIRS ${INNER_INCLUDE_DIRS})
target_include_directories(oblogmsg_base INTERFACE ${INNER_INCLUDE_DIRS})
target_compile_features(oblogmsg_base INTERFACE cxx_std_11)
add_subdirectory(src)
if (TEST)
execute_process(
COMMAND git submodule init
COMMAND git submodule update
WORKING_DIRECTORY ${PROJECT_SOURCE_DIR}
)
add_subdirectory(third_party/googletest)
add_subdirectory(unittest)
enable_testing()
endif ()
if (BENCHMARK)
add_subdirectory(benchmark)
endif ()
install(TARGETS oblogmsg oblogmsg_static
LIBRARY DESTINATION lib64
ARCHIVE DESTINATION lib64
)
install(
DIRECTORY include/ DESTINATION include/logmessage
)
benchmark/CMakeLists.txt 0 → 100644
浏览文件 @ bf5d80ed
set(SRC_LIST
lmTest.cpp
itoa.cpp
)
add_executable(lmTest ${SRC_LIST})
target_link_libraries(lmTest PUBLIC oblogmsg)
target_compile_options(lmTest PRIVATE -DLMB)
benchmark/itoa.cpp 0 → 100644
浏览文件 @ bf5d80ed
#include <cstdint>
static uint16_t MAT[] = {0x3030,
0x3130,
0x3230,
0x3330,
0x3430,
0x3530,
0x3630,
0x3730,
0x3830,
0x3930,
0x3031,
0x3131,
0x3231,
0x3331,
0x3431,
0x3531,
0x3631,
0x3731,
0x3831,
0x3931,
0x3032,
0x3132,
0x3232,
0x3332,
0x3432,
0x3532,
0x3632,
0x3732,
0x3832,
0x3932,
0x3033,
0x3133,
0x3233,
0x3333,
0x3433,
0x3533,
0x3633,
0x3733,
0x3833,
0x3933,
0x3034,
0x3134,
0x3234,
0x3334,
0x3434,
0x3534,
0x3634,
0x3734,
0x3834,
0x3934,
0x3035,
0x3135,
0x3235,
0x3335,
0x3435,
0x3535,
0x3635,
0x3735,
0x3835,
0x3935,
0x3036,
0x3136,
0x3236,
0x3336,
0x3436,
0x3536,
0x3636,
0x3736,
0x3836,
0x3936,
0x3037,
0x3137,
0x3237,
0x3337,
0x3437,
0x3537,
0x3637,
0x3737,
0x3837,
0x3937,
0x3038,
0x3138,
0x3238,
0x3338,
0x3438,
0x3538,
0x3638,
0x3738,
0x3838,
0x3938,
0x3039,
0x3139,
0x3239,
0x3339,
0x3439,
0x3539,
0x3639,
0x3739,
0x3839,
0x3939};
int I2A_64(uint64_t i, char* a)
{
if (i < (uint64_t)1000000000000) {
if (i < 100000000) {
uint32_t _i = i;
if (_i < 10) {
*a = _i + '0';
return 1;
} else if (_i < 100) {
*(uint16_t*)a = MAT[_i];
return 2;
} else if (_i < 10000) {
uint16_t A = _i / 100, B = _i - A * 100;
if (i < 1000) {
*a = A + '0';
*(uint16_t*)&a[1] = MAT[B];
return 3;
} else {
*(uint16_t*)&a[2] = MAT[B];
*(uint16_t*)a = MAT[A];
return 4;
}
} else {
uint16_t A = _i / 10000, B = _i - A * 10000, C = B / 100, D = B - C * 100;
if (A >= 100) {
/*12345678*/
if (A >= 1000) {
*(uint16_t*)a = MAT[A / 100];
*(uint16_t*)&a[2] = MAT[A % 100];
*(uint16_t*)&a[4] = MAT[C];
*(uint16_t*)&a[6] = MAT[D];
return 8;
} else {
*a = A / 100 + '0';
*(uint16_t*)&a[1] = MAT[A % 100];
*(uint16_t*)&a[3] = MAT[C];
*(uint16_t*)&a[5] = MAT[D];
return 7;
}
} else {
if (A >= 10) {
*(uint16_t*)a = MAT[A];
*(uint16_t*)&a[2] = MAT[C];
*(uint16_t*)&a[4] = MAT[D];
return 6;
} else {
*a = A + '0';
*(uint16_t*)&a[1] = MAT[C];
*(uint16_t*)&a[3] = MAT[D];
return 5;
}
}
}
} else {
uint8_t len;
uint32_t A = i / 100000000;
if (A < 10) {
*a = A + '0';
len = 1;
} else if (A < 100) {
*(uint16_t*)a = MAT[A];
len = 2;
} else if (A < 1000) {
uint16_t B = ((uint16_t)A) / 100;
*a = B + '0';
*(uint16_t*)(a + 1) = MAT[A - B * 100];
len = 3;
} else {
uint16_t B = ((uint16_t)A) / 100;
*(uint16_t*)a = MAT[B];
*(uint16_t*)(a + 2) = MAT[A - B * 100];
len = 4;
}
uint32_t _A = i - A * 100000000;
char* p = a + len;
uint16_t C;
uint16_t H = _A / 10000, L = _A - 10000 * H;
C = H / 100;
*(uint16_t*)p = MAT[C];
*(uint16_t*)(p + 2) = MAT[H - 100 * C];
C = L / 100;
*(uint16_t*)(p + 4) = MAT[C];
*(uint16_t*)(p + 6) = MAT[L - 100 * C];
return len + 8;
}
}
// 64bit max is 1844 6744 0737 0955 1615
else {
uint32_t H = i / (uint64_t)1000000000000;
unsigned char l = 0;
if (H < 100) {
if (H < 10) {
*a = H + '0';
l = 1;
} else {
*(uint16_t*)a = MAT[H];
l = 2;
}
} else if (H < 10000) {
if (H < 1000) {
*a = H / 100 + '0';
*(uint16_t*)(a + 1) = MAT[H % 100];
l = 3;
} else {
*(uint16_t*)(a) = MAT[H / 100];
*(uint16_t*)(a + 2) = MAT[H % 100];
l = 4;
}
} else if (H < 1000000) {
if (H < 100000) {
uint16_t v = H / 10000;
*a = v + '0';
v = H - 10000 * v;
*(uint16_t*)(a + 1) = MAT[v / 100];
*(uint16_t*)(a + 3) = MAT[v % 100];
l = 5;
} else {
uint16_t v = H / 10000;
*(uint16_t*)(a) = MAT[v];
v = H - 10000 * v;
*(uint16_t*)(a + 2) = MAT[v / 100];
*(uint16_t*)(a + 4) = MAT[v % 100];
l = 6;
}
} else {
if (H < 10000000) {
uint16_t v = H / 10000;
*a = v / 100 + '0';
*(uint16_t*)(a + 1) = MAT[v % 100];
v = H - 10000 * v;
*(uint16_t*)(a + 3) = MAT[v / 100];
*(uint16_t*)(a + 5) = MAT[v % 100];
l = 7;
} else {
uint16_t v = H / 10000;
*(uint16_t*)(a) = MAT[v / 100];
*(uint16_t*)(a + 2) = MAT[v % 100];
v = H - 10000 * v;
*(uint16_t*)(a + 4) = MAT[v / 100];
*(uint16_t*)(a + 6) = MAT[v % 100];
l = 8;
}
}
uint64_t L = i - (uint64_t)1000000000000 * H;
/* if(L==0)
{
memcpy(a+l,0,12);
return l+12;
}*/
uint16_t v = L / 100000000;
char* p = a + l;
uint16_t v1 = v / 100;
((uint16_t*)(p))[0] = MAT[v1];
((uint16_t*)p)[1] = MAT[v - 100 * v1];
uint32_t L1 = L - ((uint64_t)100000000) * v;
v = L1 / 10000;
v1 = v / 100;
((uint16_t*)p)[2] = MAT[v1];
((uint16_t*)p)[3] = MAT[v - 100 * v1];
v = L1 - v * 10000;
v1 = v / 100;
((uint16_t*)p)[4] = MAT[v1];
((uint16_t*)p)[5] = MAT[v - 100 * v1];
return l + 12;
}
}
int I2A_32(uint32_t i, char* a)
{
if (i < 100000000) {
uint32_t _i = i;
if (_i < 10) {
*a = _i + '0';
return 1;
} else if (_i < 100) {
*(uint16_t*)a = MAT[_i];
return 2;
} else if (_i < 10000) {
uint16_t A = _i / 100, B = _i - A * 100;
if (i < 1000) {
*a = A + '0';
*(uint16_t*)&a[1] = MAT[B];
return 3;
} else {
*(uint16_t*)&a[2] = MAT[B];
*(uint16_t*)a = MAT[A];
return 4;
}
} else {
uint16_t A = _i / 10000, B = _i - A * 10000, C = B / 100, D = B - C * 100;
if (A >= 100) {
/*12345678*/
if (A >= 1000) {
*(uint16_t*)a = MAT[A / 100];
*(uint16_t*)&a[2] = MAT[A % 100];
*(uint16_t*)&a[4] = MAT[C];
*(uint16_t*)&a[6] = MAT[D];
return 8;
} else {
*a = A / 100 + '0';
*(uint16_t*)&a[1] = MAT[A % 100];
*(uint16_t*)&a[3] = MAT[C];
*(uint16_t*)&a[5] = MAT[D];
return 7;
}
} else {
if (A >= 10) {
*(uint16_t*)a = MAT[A];
*(uint16_t*)&a[2] = MAT[C];
*(uint16_t*)&a[4] = MAT[D];
return 6;
} else {
*a = A + '0';
*(uint16_t*)&a[1] = MAT[C];
*(uint16_t*)&a[3] = MAT[D];
return 5;
}
}
}
} else {
uint8_t len;
uint32_t A = i / 100000000;
if (A < 10) {
*a = A + '0';
len = 1;
} else {
*(uint16_t*)a = MAT[A];
len = 2;
}
uint32_t _A = i - A * 100000000;
char* p = a + len;
uint16_t C;
uint16_t H = _A / 10000, L = _A - 10000 * H;
C = H / 100;
*(uint16_t*)p = MAT[C];
*(uint16_t*)(p + 2) = MAT[H - 100 * C];
C = L / 100;
*(uint16_t*)(p + 4) = MAT[C];
*(uint16_t*)(p + 6) = MAT[L - 100 * C];
return len + 8;
}
}
int I2A_16(uint16_t i, char* a)
{
if (i < 10) {
*a = i + '0';
return 1;
} else if (i < 100) {
*(uint16_t*)a = MAT[i];
return 2;
} else if (i < 10000) {
uint16_t A = i / 100, B = i - A * 100;
if (i < 1000) {
*a = A + '0';
*(uint16_t*)&a[1] = MAT[B];
return 3;
} else {
*(uint16_t*)&a[2] = MAT[B];
*(uint16_t*)a = MAT[A];
return 4;
}
} else {
uint8_t A = i / 10000;
uint16_t B = i - 10000 * A;
*a = A + '0';
*(uint16_t*)&a[1] = MAT[B / 100];
*(uint16_t*)&a[3] = MAT[B % 100];
return 5;
}
}
benchmark/itoa.h 0 → 100644
浏览文件 @ bf5d80ed
/*
* itoa.h
*
* Created on: 2015��8��18��
* Author: liwei
*/
#ifndef _ITOA_H_
#define _ITOA_H_
#include <cstdint>
int I2A_16(uint16_t i, char* a);
int I2A_32(uint32_t i, char* a);
int I2A_64(uint64_t i, char* a);
#define sitoa(i, a) ((i >= 0) ? I2A_16(i, a) : (a[0] = '-', 1 + I2A_16(~(i) + 1, a + 1)))
#define sutoa(i, a) I2A_16(i, a)
#define itoa(i, a) ((i >= 0) ? I2A_32(i, a) : (a[0] = '-', 1 + I2A_32(~(i) + 1, a + 1)))
#define utoa(i, a) I2A_32(i, a)
#define ltoa(i, a) ((i >= 0) ? I2A_64(i, a) : (a[0] = '-', 1 + I2A_64(~(i) + 1, a + 1)))
#define ultoa(i, a) I2A_64(i, a)
#endif /* _ITOA_H_ */
benchmark/lmTest.cpp 0 → 100644
浏览文件 @ bf5d80ed
此差异已折叠。
deps/logmessage.el7.x86_64.deps 0 → 100644
浏览文件 @ bf5d80ed
[target]
os=7
arch=x86_64
repo=http://mirrors.aliyun.com/oceanbase/development-kit/el/7/x86_64/
[deps]
obdevtools-cmake-3.20.2-3.el7.x86_64.rpm
obdevtools-gcc-5.2.0-3.el7.x86_64.rpm
deps/logmessage.el8.x86_64.deps 0 → 100644
浏览文件 @ bf5d80ed
[target]
os=8
arch=x86_64
repo=http://mirrors.aliyun.com/oceanbase/development-kit/el/8/x86_64/
[deps]
obdevtools-cmake-3.20.2-3.el8.x86_64.rpm
obdevtools-gcc-5.2.0-3.el8.x86_64.rpm
include/BinLogBuf.h 0 → 100644
浏览文件 @ bf5d80ed
#ifndef BINLOG_BUF_H_
#define BINLOG_BUF_H_
#define LOG_EVENT_INIT_LEN (1024 * 32)
#include <cstdlib>
#include <cstring>
namespace oceanbase {
namespace logmessage {
struct BinLogBuf {
char* buf;
size_t buf_size;
size_t buf_used_size;
bool needFree;
BinLogBuf()
{
buf = NULL;
buf_size = buf_used_size = 0;
needFree = false;
}
BinLogBuf(char* binlog, size_t len)
{
buf = binlog;
buf_size = len;
buf_used_size = 0;
needFree = false;
}
~BinLogBuf()
{
if (buf != NULL && needFree == true)
delete[] buf;
}
};
static inline int set_binlogBuf(BinLogBuf* buf, char* binlog, size_t len)
{
bool needRealloc = true;
if (buf->buf == NULL) {
len > LOG_EVENT_INIT_LEN ? (buf->buf_size = len) : (buf->buf_size = LOG_EVENT_INIT_LEN);
} else if (buf->buf_size < len) {
buf->buf_size = len;
delete[] buf->buf;
} else if (LOG_EVENT_INIT_LEN > len && buf->buf_size > LOG_EVENT_INIT_LEN) {
buf->buf_size = LOG_EVENT_INIT_LEN;
delete[] buf->buf;
} else
needRealloc = false;
if (needRealloc) {
if ((buf->buf = new char[buf->buf_size]) == NULL)
return -1;
}
buf->needFree = true;
memcpy(buf->buf, binlog, len);
buf->buf_used_size = len;
return 0;
}
static inline void exchange_binlogBuf(BinLogBuf* buf, char* binlog, size_t len, size_t buf_used_size = 0)
{
if (buf->needFree == true)
delete[] buf->buf;
buf->buf = binlog;
buf->buf_size = len;
buf->buf_used_size = buf_used_size;
buf->needFree = false;
}
static inline void exchange_binlogBuf(BinLogBuf* buf, BinLogBuf* ebuf)
{
if (buf->needFree == true)
delete[] buf->buf;
if (ebuf == NULL) {
buf->buf = NULL;
buf->buf_size = buf->buf_used_size = 0;
return;
}
buf->buf = ebuf->buf;
ebuf->buf = NULL;
buf->buf_size = ebuf->buf_size;
ebuf->buf_size = 0;
buf->buf_used_size = ebuf->buf_used_size;
ebuf->buf_used_size = 0;
buf->needFree = ebuf->needFree;
ebuf->needFree = false;
}
static inline void get_binlogBuf(BinLogBuf* buf, char* binlog, size_t len)
{
if (buf->needFree == true)
delete[] buf->buf;
buf->buf = binlog;
buf->buf_used_size = buf->buf_size = len;
buf->needFree = false;
}
static inline int create_binlogBuf(BinLogBuf* buf, size_t len = LOG_EVENT_INIT_LEN)
{
if (NULL == (buf->buf = new char[buf->buf_size = len]))
return -1;
buf->needFree = true;
return (int)(buf->buf_used_size = 0);
}
static inline void clear_binlogBuf(BinLogBuf* buf)
{
buf->buf_used_size = 0;
}
static inline bool binlogBuf_NULL(BinLogBuf* buf)
{
return buf->buf == NULL;
}
} // namespace logmessage
} // namespace oceanbase
#endif /* BINLOG_BUF_H_ */
include/Endian.h 0 → 100644
浏览文件 @ bf5d80ed
#ifndef ENDIANNESS_H
#define ENDIANNESS_H
namespace oceanbase {
namespace logmessage {
/* This should catch all modern GCCs and Clang */
#if (defined __BYTE_ORDER__) && (defined __ORDER_LITTLE_ENDIAN__)
#if __BYTE_ORDER__ == __ORDER_LITTLE_ENDIAN__
#define ENDIANNESS_LE 1
#define ENDIANNESS_BE 0
#elif __BYTE_ORDER__ == __ORDER_BIG_ENDIAN__
#define ENDIANNESS_LE 0
#define ENDIANNESS_BE 1
#endif
/* Try to derive from arch/compiler-specific macros */
#elif defined(_X86_) || defined(__x86_64__) || defined(__i386__) || defined(__i486__) || defined(__i586__) || \
defined(__i686__) || defined(__MIPSEL) || defined(_MIPSEL) || defined(MIPSEL) || defined(__ARMEL__) || \
(defined(__LITTLE_ENDIAN__) && __LITTLE_ENDIAN__ == 1) || (defined(_LITTLE_ENDIAN) && _LITTLE_ENDIAN == 1) || \
defined(_M_IX86) || defined(_M_AMD64) /* MSVC */
#define ENDIANNESS_LE 1
#define ENDIANNESS_BE 0
#elif defined(__MIPSEB) || defined(_MIPSEB) || defined(MIPSEB) || defined(__MICROBLAZEEB__) || defined(__ARMEB__) || \
(defined(__BIG_ENDIAN__) && __BIG_ENDIAN__ == 1) || (defined(_BIG_ENDIAN) && _BIG_ENDIAN == 1)
#define ENDIANNESS_LE 0
#define ENDIANNESS_BE 1
/* Try to get it from a header */
#else
#if defined(__linux)
#include <endian.h>
#else
#include <machine/endian.h>
#endif
#endif
#ifndef ENDIANNESS_LE
#undef ENDIANNESS_BE
#if defined(__BYTE_ORDER) && defined(__LITTLE_ENDIAN)
#if __BYTE_ORDER == __LITTLE_ENDIAN
#define ENDIANNESS_LE 1
#define ENDIANNESS_BE 0
#elif __BYTE_ORDER == __BIG_ENDIAN
#define ENDIANNESS_LE 0
#define ENDIANNESS_BE 1
#endif
#elif defined(BYTE_ORDER) && defined(LITTLE_ENDIAN)
#if BYTE_ORDER == LITTLE_ENDIAN
#define ENDIANNESS_LE 1
#define ENDIANNESS_BE 0
#elif BYTE_ORDER == BIG_ENDIAN
#define ENDIANNESS_LE 0
#define ENDIANNESS_BE 1
#endif
#endif
#endif
/* In case the user passed one of -DENDIANNESS_LE or BE in CPPFLAS,
* set the second one too */
#if defined(ENDIANNESS_LE) && !(defined(ENDIANNESS_BE))
#if ENDIANNESS_LE == 0
#define ENDIANNESS_BE 1
#else
#define ENDIANNESS_BE 0
#endif
#elif defined(ENDIANNESS_BE) && !(defined(ENDIANNESS_LE))
#if ENDIANNESS_BE == 0
#define ENDIANNESS_LE 1
#else
#define ENDIANNESS_LE 0
#endif
#endif
#if !(defined(ENDIANNESS_LE)) && !(defined(ENDIANNESS_PORTABLE_CONVERSION))
#error \
"Sorry, we couldn't detect endiannes for your system! Please set -DENDIANNESS_LE=1 or 0 using your CPPFLAGS/CFLAGS!"
#endif
/*little endian set and get*/
static inline uint8_t swap8(uint8_t __x)
{
return __x;
}
static inline uint16_t swap16(uint16_t __x)
{
return (__x << 8) | (__x >> 8);
}
static inline uint32_t swap32(uint32_t __x)
{
return (__x >> 24) | (__x >> 8 & 0xff00) | (__x << 8 & 0xff0000) | (__x << 24);
}
static inline uint64_t swap64(uint64_t __x)
{
return ((swap32((uint32_t)__x) + 0ULL) << 32) | (swap32(__x >> 32));
}
static inline void toLeEndian(void* buf, int size)
{
#if ENDIANNESS_LE
return;
#endif
switch (size) {
case 1:
*((uint8_t*)buf) = swap8(*(uint8_t*)buf);
break;
case 2:
*((uint16_t*)buf) = swap16(*(uint16_t*)buf);
break;
case 4:
*((uint32_t*)buf) = swap32(*(uint32_t*)buf);
break;
case 8:
*((uint64_t*)buf) = swap64(*(uint64_t*)buf);
break;
}
}
template <class T>
static inline T toLeEndianByType(T value)
{
#if ENDIANNESS_LE
return value;
#endif
toLeEndian(&value, sizeof(T));
return value;
}
} // namespace logmessage
} // namespace oceanbase
#endif
include/LogMsgBuf.h 0 → 100644
浏览文件 @ bf5d80ed
#ifndef _LOGMSGBUF_H_
#define _LOGMSGBUF_H_
#include <vector>
#include <string>
using namespace std;
namespace oceanbase {
namespace logmessage {
struct BinLogBuf;
int LogMsgInit();
int LogMsgLocalInit();
void LogMsgDestroy();
void LogMsgLocalDestroy();
const char* LogMsgGetValueByOffset(size_t offset);
size_t LogMsgAppendString(const char* string, size_t size);
size_t LogMsgAppendString(const std::string& string);
size_t LogMsgAppendBuf(const char* data, size_t size);
size_t LogMsgAppendBuf(const std::string& string);
size_t LogMsgAppendBuf(const BinLogBuf* sa, size_t size);
size_t LogMsgAppendStringArray(std::vector<std::string*>& sa);
size_t LogMsgAppendStringArray(const char** sa, size_t size);
void LogMsgSetHead(size_t size);
void LogMsgCopyHead(const char* head, size_t size);
void LogMsgFroceSetHeadSize(size_t size);
const char* LogMsgGetString(size_t* size);
size_t LogMsgAppendDataArray(std::vector<long>& sa);
size_t LogMsgAppendDataArray(uint8_t* sa, size_t size);
} // namespace logmessage
} // namespace oceanbase
#endif /* _LOGMSGBUF_H_ */
include/LogMsgFactory.h 0 → 100644
浏览文件 @ bf5d80ed
#ifndef __LOG_MSG_FACTORY__
#define __LOG_MSG_FACTORY__
#include <string>
namespace oceanbase {
namespace logmessage {
class IColMeta;
class ITableMeta;
class IDBMeta;
class IMetaDataCollections;
class ILogRecord;
class LogMsgFactory {
public:
static std::string DFT_ColMeta;
static std::string DFT_TableMeta;
static std::string DFT_DBMeta;
static std::string DFT_METAS;
static std::string DFT_LR;
public:
/**
* Create an empty ColMeta by a specific type
*
* @param type is the name of the specific colmeta, default is "ColMetaImpl"
* @return instantialized colmeta
*/
static IColMeta* createColMeta(const std::string& type = DFT_ColMeta);
/**
* Create a meaningful ColMeta by a specific type with given serialized
* data and size
*
* @param type is the name of the specific colmeta, default is "ColMetaImpl"
* @param ptr is the address of a consecutive memory in which the
* serialized colmeta is stored
* @param size is the length of the consecutive memory
* @return instantialized colmeta
*/
static IColMeta* createColMeta(const std::string& type, const char* ptr, size_t size);
/**
* Create an empty TableMeta by a specific type
*
* @param type is the name of the specific tablemeta, default is "TableMetaImpl"
* @return instantialized tablemeta
*/
static ITableMeta* createTableMeta(const std::string& type = DFT_TableMeta);
/**
* Create a meaningful TableMeta by a specific type with given serialized data
* and size
*
* @param type is the name of the specific tablemeta, default is
* "TableMetaImpl"
* @param ptr is the address of a consecutive memory in which the
* date of a serialized tablemeta is stored
* @param size is the length of the consecutive memory
* @return instantialized tablemeta
*/
static ITableMeta* createTableMeta(const std::string& type, const char* ptr, size_t size);
/**
* Create an empty DBMeta by a specific type
*
* @param type is the name of the specific dbmeta, default is "DBMetaImpl"
* @return instantialized dbmeta
*/
static IDBMeta* createDBMeta(const std::string& type = DFT_DBMeta);
/**
* Create a meaningful DBMeta by a specific type with given serialized
* data and size
*
* @param type is the name of the specific dbmeta, default is "DBMetaImpl"
* @param ptr is the address of a consecutive memory in which the
* serialized dbmeta is stored
* @param size is the length of the consecutive memory
* @return instantialized dbmeta
*/
static IDBMeta* createDBMeta(const std::string& type, const char* ptr, size_t size);
/**
* Create an empty MetaDataCollections by a specific type
*
* @param type is the name of the specific metadatacollections,
* default is "MetaDataCollectionsImpl"
* @return instantialized metadatacollections
*/
static IMetaDataCollections* createMetaDataCollections(const std::string& type = DFT_METAS);
/**
* Create a meaningful MetaDataCollections by a specific type with given
* serialized data and size
*
* @param type is the name of the specific metadatacollections,
* default is "DBMetaImpl"
* @param ptr is the address of a consecutive memory in which the
* serialized metadatacollections is stored
* @param size is the length of the consecutive memory
* @param freeMem is to direct whether the assigned ptr should be
* freed after the metadatacollections is destroyed
* @return instantialized metadatacollections
*/
static IMetaDataCollections* createMetaDataCollections(
const std::string& type, const char* ptr, size_t size, bool freeMem);
/**
* Create a binlog record by the specific type with the judgement whether
* need initilizing non-serialized members in memory
*
* @param type is the name of the specific LogRecord,
* default is "LogRecordImpl"
* @param creating is to differentiate two kinds of usage, if creating is
* true, it means the created binlog record has not been
* serilized, all in-memory functions can be called. Otherwise
* if creating is false, only after-serialized function could
* be called
* @return the instantialized binlog record
*/
static ILogRecord* createLogRecord(const std::string& type = DFT_LR, bool creating = true);
/**
* Create a binlog record by the specific type from data stored in the
* consecutive memory directed by ptr with length size
*
* @param type is the name of the specific LogRecord,
* default is "LogRecordImpl"
* @param ptr is the address of the consecutive memory
* @param size is the length of the memory
* @return the instantialized binlog record
*/
static ILogRecord* createLogRecord(const std::string& type, const char* ptr, size_t size);
/**
* Destroy the binlog record created by the factory with the memory
* allocated by users with putOld and putNew
*
* @param record is the created record which is freed hereby
*/
static void destroy(ILogRecord*& record);
/**
* Destroy the binlog record created by the factory, the users memory
* is not freed
*
* @param record is the created record which is freed hereby
*/
static void destroyWithUserMemory(ILogRecord*& record);
/**
* Destroy the column meta created by the factory
*
* @param colMeta is the meta of the column to be freed
*/
static void destroy(IColMeta*& colMeta);
/**
* Destroy the table meta created by the factory
*
* @param tableMeta is the meta of the table to be freed
*/
static void destroy(ITableMeta*& tableMeta);
/**
* Destroy the database meta created by the factory
*
* @param dbMeta is the meta of the database to be freed
*/
static void destroy(IDBMeta*& dbMeta);
/**
* Destroy the collections of all db meta created by the factory
*
* @param metaColls is the collections of all meta to be freed
*/
static void destroy(IMetaDataCollections*& metaColls);
};
} // namespace logmessage
} // namespace oceanbase
#endif
include/LogRecord.h 0 → 100644
浏览文件 @ bf5d80ed
#ifndef _LOG_REC_H_
#define _LOG_REC_H_
#include <string>
#include <vector>
#include <cstdint>
#include "UserData.h"
#include "BinLogBuf.h"
namespace oceanbase {
namespace logmessage {
class StrArray;
struct LogRecInfo;
#define BR_FAKE_DDL_COLNAME "ddl"
typedef enum RecordType {
EINSERT = 0,
EUPDATE,
EDELETE,
EREPLACE,
HEARTBEAT,
CONSISTENCY_TEST,
EBEGIN,
ECOMMIT,
EDDL,
EROLLBACK,
EDML,
INDEX_INSERT = 128,
INDEX_UPDATE,
INDEX_DELETE,
INDEX_REPLACE,
EUNKNOWN = 255,
} RECORD_TYPE;
enum FORMATERR {
FORMATOK = 0,
BUFOVERFLOW,
TYPENOTSUPPORT,
UNKNOWNTYPE,
UNKNOWNOPTYPE,
NULLTBLMETA,
NULLCOLMETA,
ENUMSETNULL
};
enum SOURCE_TYPE {
SRC_MYSQL = 0x00, // MySQL
SRC_OCEANBASE = 0x01, // Oceanbase
SRC_HBASE = 0X02, // HBase
SRC_ORACLE = 0x03, // Oracle
SRC_OCEANBASE_1_0 = 0x04, // Oceanbase V1
SRC_DB2 = 0x05, // DB2
SRC_UNKNOWN = 0x06
};
enum SOURCE_CATEGORY {
SRC_FULL_RECORDED = 0, // can get all column value from redo log
SRC_FULL_RETRIEVED, // can get part column value from redo log,other need fetch
SRC_FULL_FAKED,
SRC_PART_RECORDED,
SRC_NO
};
class ITableMeta;
class IDBMeta;
class IMetaDataCollections;
class ILogRecord : public UserDataInterface {
public:
virtual ~ILogRecord()
{}
public:
/* setter and getter */
// must set: record src type
virtual void setSrcType(int type) = 0;
virtual int getSrcType() const = 0;
// must set
virtual void setSrcCategory(int cartegory) = 0;
virtual int getSrcCategory() const = 0;
virtual void setThreadId(uint32_t threadId) = 0;
virtual uint32_t getThreadId() = 0;
// must set: record's create time
virtual void setTimestamp(long timestamp) = 0;
virtual time_t getTimestamp() = 0;
// must set: record's type,like insert/delete/update/replace/heartbeat...
virtual int setRecordType(int aType) = 0;
virtual int recordType() = 0;
// must set: ITableMeta's encoding
virtual const char* recordEncoding() = 0;
// must set: database name
virtual void setDbname(const char* db) = 0;
virtual const char* dbname() const = 0;
// must set: table name
virtual void setTbname(const char* table) = 0;
virtual const char* tbname() const = 0;
// musst set when creating mode, parse mode no need: table's meta info
virtual void setTableMeta(ITableMeta* tblMeta) = 0;
virtual int getTableMeta(ITableMeta*& tblMeta) = 0;
// return table meta directly
virtual ITableMeta* getTableMeta() = 0;
// is record parsed from serialized data
virtual bool isParsedRecord() = 0;
virtual void setDBMeta(IDBMeta* dbMeta) = 0;
virtual IDBMeta* getDBMeta() = 0;
virtual void setExpiredTableMeta(ITableMeta* tblMeta) = 0;
virtual std::vector<ITableMeta*>& getExpiredTableMeta() = 0;
virtual void setExpiredDBMeta(IDBMeta* dbMeta) = 0;
virtual std::vector<IDBMeta*>& getExpiredDBMeta() = 0;
virtual void setExpiredMetaDataCollections(IMetaDataCollections* imc) = 0;
virtual IMetaDataCollections* getExpiredMetaDataCollections() = 0;
// muset set: set checkpoint,return like %d@%d
virtual void setCheckpoint(uint64_t file, uint64_t offset) = 0;
virtual const char* getCheckpoint() = 0;
/**
* ob: getCheckpoint1 High 32 bit checkpoint, getCheckpoint2 Low 32 bit checkpoint
*/
virtual uint64_t getCheckpoint1() = 0;
virtual uint64_t getCheckpoint2() = 0;
virtual uint64_t getFileNameOffset() = 0;
virtual uint64_t getFileOffset() = 0;
virtual void setFirstInLogevent(bool b) = 0;
virtual bool firstInLogevent() = 0;
virtual void setId(uint64_t id) = 0;
virtual uint64_t id() = 0;
virtual void setInstance(const char* instance) = 0;
virtual const char* instance() const = 0;
// this value will save at pkValue
virtual void setExtraInfo(const char* info) = 0;
virtual StrArray* extraInfo() const = 0;
virtual bool isTimemarked() const = 0;
virtual void setTimemarked(bool marked) = 0;
virtual void addTimemark(long time) = 0;
virtual std::vector<long>& getTimemark() = 0;
virtual std::vector<long>& getTimemark(size_t& length) = 0;
virtual void curveTimemark() = 0;
public:
/* column data */
/**
* 1. insert no old value,only has new value
* 2. delete no new value,only has old value
* 3. update/replace has new and old value
* 4. column must sort by ITableMeta's column
*/
virtual int putOld(std::string* val) = 0;
virtual int putNew(std::string* val) = 0;
virtual int putOld(const char* pos, int len) = 0;
virtual int putNew(const char* pos, int len) = 0;
virtual void setNewColumn(BinLogBuf* buf, int size) = 0;
virtual void setOldColumn(BinLogBuf* buf, int size) = 0;
virtual int getColumnCount() = 0;
virtual int putFilterRuleVal(const char* pos, int len) = 0;
virtual int16_t getRecordHash(
int16_t (*hashFunc)(const char** valueList, const size_t* valueSizeList, int count)) = 0;
virtual bool hashColumnSetted() = 0;
virtual void setHashFuncId(int id) = 0;
virtual int getHashFuncId() = 0;
virtual int setHashCol(std::vector<std::string>& hashColumns) = 0;
virtual void setHashColByPK() = 0;
virtual void clearOld() = 0;
virtual void clearNew() = 0;
// get all old value
virtual const std::vector<std::string*>& oldCols() = 0;
virtual BinLogBuf* oldCols(unsigned int& count) = 0;
virtual StrArray* parsedOldCols() const = 0;
// get all new value
virtual const std::vector<std::string*>& newCols() = 0;
virtual BinLogBuf* newCols(unsigned int& count) = 0;
virtual StrArray* parsedNewCols() const = 0;
// get all column names
virtual StrArray* parsedColNames() const = 0;
virtual StrArray* parsedColEncodings() const = 0;
virtual StrArray* parsedFilterRuleValues() const = 0;
virtual const uint8_t* parsedColTypes() const = 0;
virtual const uint8_t* parsedColFlags() const = 0;
virtual const std::vector<int>& pkKeys() = 0;
virtual const std::vector<int>& ukKeys() = 0;
public:
/* tostring and parse */
virtual int parse(const void* ptr, size_t size) = 0;
virtual int parseFast(const void* ptr, size_t size) = 0;
/**
* @return true success; false fail
*/
virtual bool parsedOK() = 0;
virtual size_t getRealSize() = 0;
/**
* serialize
* @return serialized ptr
*/
virtual const char* toString(size_t* size, bool reserveMemory = false) = 0;
/**
* @return serialized ptr
*/
virtual const char* getFormatedString(size_t* size) = 0;
/**
* clear all to reuse, no free memory
*/
virtual void clear() = 0;
/**
* clear all,include memory
*/
virtual void clearWithUserMemory() = 0;
virtual void clearExpiredMeta() = 0;
virtual void setRecordEncoding(const char* encoding) = 0;
virtual void setSqlNo(int32_t sql_no) = 0;
virtual int32_t sqlNo() = 0;
virtual void setObTraceInfo(const char* ob_trace_info) = 0;
virtual const char* obTraceInfo() = 0;
// must set: record's create us
virtual void setRecordUsec(uint32_t usec) = 0;
virtual uint32_t getRecordUsec() = 0;
// get serialized string of record directly
virtual const char* getSerializedString(size_t* size) = 0;
};
class LogRecordImpl : public ILogRecord {
public:
LogRecordImpl(time_t timestamp, ITableMeta* tblMeta);
LogRecordImpl(const void* ptr, size_t size);
LogRecordImpl(bool creating = true, bool useDMB = false);
virtual ~LogRecordImpl();
public:
/* setter and getter */
virtual void setSrcType(int type);
virtual int getSrcType() const;
virtual void setSrcCategory(int cartegory);
virtual int getSrcCategory() const;
virtual void setTimestamp(long timestamp);
virtual time_t getTimestamp();
virtual int setRecordType(int aType);
virtual int recordType();
virtual const char* recordEncoding();
virtual void setDbname(const char* db);
virtual const char* dbname() const;
virtual void setTbname(const char* table);
virtual const char* tbname() const;
virtual void setTableMeta(ITableMeta* tblMeta);
// if record is parsed from serialized data, tblMeta can not be null
// and will be set values parsed from serialized data.
// if tblMeta is Null pointer, will set current table meta to tblMeta
virtual int getTableMeta(ITableMeta*& tblMeta);
// return table meta directly
virtual ITableMeta* getTableMeta();
// is record parsed from serialized data
virtual bool isParsedRecord();
virtual void setDBMeta(IDBMeta* dbMeta);
virtual IDBMeta* getDBMeta();
virtual void setExpiredTableMeta(ITableMeta* tblMeta);
virtual std::vector<ITableMeta*>& getExpiredTableMeta();
virtual void setExpiredDBMeta(IDBMeta* dbMeta);
virtual std::vector<IDBMeta*>& getExpiredDBMeta();
virtual void setExpiredMetaDataCollections(IMetaDataCollections* imc);
virtual IMetaDataCollections* getExpiredMetaDataCollections();
virtual void setCheckpoint(uint64_t file, uint64_t offset);
virtual const char* getCheckpoint();
virtual uint64_t getCheckpoint1();
virtual uint64_t getCheckpoint2();
virtual uint64_t getFileNameOffset();
virtual uint64_t getFileOffset();
virtual void setFirstInLogevent(bool b);
virtual bool firstInLogevent();
virtual void setRecordUsec(uint32_t usec);
virtual uint32_t getRecordUsec();
virtual void setId(uint64_t id);
virtual uint64_t id();
virtual void setInstance(const char* instance);
virtual const char* instance() const;
virtual void setExtraInfo(const char* info);
virtual StrArray* extraInfo() const;
bool isTimemarked() const;
void setTimemarked(bool marked);
void addTimemark(long time);
std::vector<long>& getTimemark();
std::vector<long>& getTimemark(size_t& length);
virtual void curveTimemark();
virtual void setThreadId(uint32_t threadId);
virtual uint32_t getThreadId();
public:
/* column data */
virtual int putOld(std::string* val);
virtual int putNew(std::string* val);
virtual int putOld(const char* pos, int len);
virtual int putNew(const char* pos, int len);
virtual void setNewColumn(BinLogBuf* buf, int size);
virtual void setOldColumn(BinLogBuf* buf, int size);
virtual int getColumnCount();
virtual int putFilterRuleVal(const char* pos, int len);
virtual void clearOld();
virtual void clearNew();
virtual const std::vector<std::string*>& oldCols();
virtual BinLogBuf* oldCols(unsigned int& count);
virtual StrArray* parsedOldCols() const;
virtual const std::vector<std::string*>& newCols();
virtual BinLogBuf* newCols(unsigned int& count);
virtual const BinLogBuf* filterValues(unsigned int& count);
virtual StrArray* parsedNewCols() const;
virtual StrArray* parsedFilterRuleValues() const;
virtual StrArray* parsedColNames() const;
virtual StrArray* parsedColEncodings() const;
virtual const uint8_t* parsedColTypes() const;
virtual const uint8_t* parsedColFlags() const;
virtual const std::vector<int>& pkKeys();
virtual const std::vector<int>& ukKeys();
void elementAtPk(int off, char*& v, size_t& size) const;
void elementAtNew(int off, char*& v, size_t& size) const;
void elementAtOld(int off, char*& v, size_t& size) const;
const char* parseColumnValue(const char* columnName, size_t* size, int* columnType);
const char* parseColumnValue(const char* columnName, size_t* size, int* columnType, bool isPre);
public:
/* tostring and parse */
virtual int parse(const void* ptr, size_t size);
virtual int parseFast(const void* ptr, size_t size);
virtual bool parsedOK();
virtual size_t getRealSize();
virtual const char* toString(size_t* size, bool reserveMemory = false);
virtual const char* getFormatedString(size_t* size);
virtual void clear();
virtual void clearWithUserMemory();
virtual void clearExpiredMeta();
virtual bool hashColumnSetted();
virtual int16_t getRecordHash(int16_t (*hashFunc)(const char** valueList, const size_t* valueSizeList, int count));
virtual void setHashFuncId(int id);
virtual int getHashFuncId();
virtual int setHashCol(std::vector<std::string>& hashColumns);
virtual void setHashColByPK();
virtual void getPKStringArrayData(const char*& v, size_t& size);
public:
virtual void setUserData(void* data);
virtual void* getUserData();
virtual void setRecordEncoding(const char* encoding);
virtual void setSqlNo(int32_t sql_no);
virtual int32_t sqlNo();
virtual void setObTraceInfo(const char* ob_trace_info);
virtual const char* obTraceInfo();
virtual const char* getSerializedString(size_t* size);
protected:
LogRecInfo* m_lr;
std::string m_buf;
bool m_timemarked;
void* m_userData;
};
} // namespace logmessage
} // namespace oceanbase
#endif /*_LOG_REC_H_*/
include/MetaInfo.h 0 → 100644
浏览文件 @ bf5d80ed
#ifndef _MD_H_
#define _MD_H_
#include <cstdio>
#include <ctime>
#include <pthread.h>
#include <vector>
#include <string>
#include "UserData.h"
namespace oceanbase {
namespace logmessage {
class StrArray;
enum logmsg_field_types {
LOGMSG_TYPE_DECIMAL,
LOGMSG_TYPE_TINY,
LOGMSG_TYPE_SHORT,
LOGMSG_TYPE_LONG,
LOGMSG_TYPE_FLOAT,
LOGMSG_TYPE_DOUBLE,
LOGMSG_TYPE_NULL,
LOGMSG_TYPE_TIMESTAMP,
LOGMSG_TYPE_LONGLONG,
LOGMSG_TYPE_INT24,
LOGMSG_TYPE_DATE,
LOGMSG_TYPE_TIME,
LOGMSG_TYPE_DATETIME,
LOGMSG_TYPE_YEAR,
LOGMSG_TYPE_NEWDATE,
LOGMSG_TYPE_VARCHAR,
LOGMSG_TYPE_BIT,
LOGMSG_TYPE_NEWDECIMAL = 246,
LOGMSG_TYPE_ENUM = 247,
LOGMSG_TYPE_SET = 248,
LOGMSG_TYPE_TINY_BLOB = 249,
LOGMSG_TYPE_MEDIUM_BLOB = 250,
LOGMSG_TYPE_LONG_BLOB = 251,
LOGMSG_TYPE_BLOB = 252,
LOGMSG_TYPE_VAR_STRING = 253,
LOGMSG_TYPE_STRING = 254,
LOGMSG_TYPE_GEOMETRY = 255,
LOGMSG_TYPES
};
// column's meta
struct ColMetaInfo;
class IDBMeta;
class IMetaDataCollections;
struct _trie_tree;
class IColMeta : public UserDataInterface {
public:
IColMeta();
IColMeta(const void* ptr, size_t size);
virtual ~IColMeta();
public:
// override
virtual const char* getName();
virtual int getType();
virtual long getLength();
virtual bool isSigned();
virtual bool isPK();
virtual bool isRuleCol();
virtual bool isUK();
virtual bool isNotNull();
virtual int getDecimals();
virtual const char* getDefault();
virtual const char* getEncoding();
virtual int getRequired();
virtual StrArray* getValuesOfEnumSet();
virtual bool isGenerated();
virtual void setFlag(unsigned char flag);
virtual unsigned char getFlag();
virtual void setName(const char* name);
virtual void setType(int type);
virtual void setLength(long length);
virtual void setSigned(bool b);
virtual void setIsPK(bool b);
virtual void setIsRuleCol(bool b);
virtual void setIsUK(bool b);
virtual void setNotNull(bool b);
virtual void setDecimals(int decimals);
virtual void setDefault(const char* def);
virtual void setDefault(const char* def, size_t length);
virtual void setEncoding(const char* enc);
virtual void setRequired(int required);
virtual void setValuesOfEnumSet(std::vector<std::string>& v);
virtual void setValuesOfEnumSet(std::vector<const char*>& v);
virtual void setValuesOfEnumSet(const char** v, size_t size);
virtual void setGenerated(bool Generated);
virtual void setHiddenRowKey();
virtual bool isHiddenRowKey();
public:
int appendTo(std::string& s);
size_t getRealSize();
int parse(const void* ptr, size_t size);
bool parsedOK();
public:
virtual void setUserData(void* data);
virtual void* getUserData();
private:
ColMetaInfo* m_col;
void* m_userData;
};
// table's meta
struct TableMetaInfo;
class ITableMeta : public UserDataInterface {
public:
ITableMeta();
ITableMeta(const void* ptr, size_t size);
virtual ~ITableMeta();
public:
// override
virtual const char* getName();
virtual bool hasPK();
virtual bool hasUK();
virtual const char* getPKs();
virtual const char* getUKs();
virtual const int* getPKs(int& size) const;
virtual const int* getUKs(int& size) const;
virtual const char* getEncoding();
virtual IDBMeta* getDBMeta();
virtual void setName(const char* name);
virtual void setHasPK(bool b);
virtual void setPKIndice(const std::vector<int>& indice);
virtual void setPKs(const char* pks);
virtual void setHasUK(bool b);
virtual void setUKs(const char* uks);
virtual void setEncoding(const char* enc);
virtual void setDBMeta(IDBMeta* dbMeta);
virtual void setPkinfo(const char* info);
virtual const char* getPkinfo();
virtual void setUkinfo(const char* info);
virtual const char* getUkinfo();
bool hashColumnSetted();
const int* getHashColumnIdx(int& hashCoumnCount, char**& hashValueList, size_t*& hashValueSizeList);
void setHashColByPK();
int setHashCol(std::vector<std::string>& hashColumns);
void setHashFuncId(int id);
int getHashFuncId();
void trySerializeMetaDataAsMsgArea(std::vector<const char*>& extra_infos);
const std::string& getNameData();
const std::string& getEncodingData();
const std::string& getcolTypeData();
const std::string& getPkData();
const std::string& getUkData();
const std::string& getkeyData();
const std::string& getColumnFlagData();
const std::string& getNotNullData();
const std::string& getSignedData();
const std::string& getDecimalsData();
const std::string& getDefaultData();
public:
/**
* get all column name
*/
std::vector<std::string>& getColNames();
/**
* get PK column name
*/
std::vector<std::string>& getPKColNames();
std::vector<std::string>& getUKColNames();
/**
* get column's meta by name
*/
virtual IColMeta* getCol(const char* colName);
virtual int getColIndex(const char* colName);
/**
* get Column count
*/
virtual int getColCount();
/*
* get column's meta by column id
*/
virtual IColMeta* getCol(int index);
virtual int getColNum(const char* colName);
/**
* append a column meta
*/
virtual int append(const char* colName, IColMeta* colMeta);
/* For partial */
virtual bool isDropped();
virtual void setDropped(bool value);
virtual const char* getNewName();
virtual void setNewName(const char* name);
public:
int appendTo(std::string& s);
size_t getRealSize();
int parse(const void* ptr, size_t size);
bool parsedOK();
/*
* parse key index like (2, 0, 1) or (0, 1, 2)
* return index vector, if format is invalid,
* return -1, if success, return 0.
*/
int parseKeyIndex(std::string indexStr, std::vector<int>& indice);
/*
* get pk or uk keys from pkinfo or ukinfo like (0, 1, 2......)
*/
int getKeysFromInfo(std::string info, std::string& keys, StrArray* colNames);
public:
virtual void setUserData(void* data);
virtual void* getUserData();
virtual void** getUserDataPtr();
private:
TableMetaInfo* m_tbl;
void* m_userData;
std::string m_colNameData;
std::string m_encodingData;
std::string m_colTypeData;
std::string m_PkData;
std::string m_UkData;
std::string m_keyData;
std::string m_columnFlagData;
std::string m_colNotNullData;
std::string m_colSignedData;
std::string m_colDecimalsData;
std::string m_colDefaultData;
pthread_mutex_t m_mdMutex;
bool m_DataOk;
int* m_hashColumnIdx;
int m_hashColumnSize;
bool m_hashColumnSetted;
int m_hashFuncId;
char** m_hashColumValueList;
size_t* m_hashColumValueSizeList;
};
// DB's meta
struct DBMetaInfo;
class IDBMeta : public UserDataInterface {
public:
IDBMeta();
IDBMeta(const void* ptr, size_t size);
virtual ~IDBMeta();
public:
// override
virtual const char* getName();
virtual const char* getEncoding();
virtual IMetaDataCollections* getMetaDataCollections();
virtual void setName(const char* name);
virtual void setEncoding(const char* enc);
virtual void setMetaDataCollections(IMetaDataCollections* mdc);
virtual int getTblCount();
/**
* get table's meta by name
*/
virtual ITableMeta* get(const char* tblName);
virtual ITableMeta* get(int index);
/**
* add a table's meta
*/
virtual int put(ITableMeta* tblMeta);
/* For partial */
virtual bool isDropped();
virtual void setDropped(bool value);
public:
virtual int eraseMapIterator();
virtual int getFromMapIterator(const char** tblName, ITableMeta** tblMeta);
virtual int nextMapIterator(bool erase);
virtual int resetMapIterator();
virtual int erase(const char* tableName, bool delayDeleteMeta);
virtual int appendTo(std::string& s);
virtual size_t getRealSize();
virtual int parse(const void* ptr, size_t size);
virtual bool parsedOK();
public:
virtual void setUserData(void* data);
virtual void* getUserData();
virtual void** getUserDataPtr();
private:
DBMetaInfo* m_db;
void* m_userData;
};
// all db's meta
struct MetaDataCollectionInfo;
class IMetaDataCollections : public UserDataInterface {
public:
IMetaDataCollections();
IMetaDataCollections(
const void* ptr, size_t size, bool removePtr = false); // removePtr means if free ptr when destroy object
virtual ~IMetaDataCollections();
public:
// override
virtual unsigned getMetaVerNum();
virtual IMetaDataCollections* getPrev();
virtual time_t getTimestamp();
virtual void setMetaVerNum(unsigned metaVerNum);
virtual void setPrev(IMetaDataCollections* prev);
virtual void setTimestamp(time_t timestamp);
/**
* get Db count
*/
virtual int getDbCount();
/**
* get Db meta by name
*/
virtual IDBMeta* get(const char* dbname);
/**
* get Db meta by id
*/
virtual IDBMeta* get(int index);
/**
* get table meta by dbname and tablename
*/
virtual ITableMeta* get(const char* dbName, const char* tblName);
/**
* add a db meta
*/
virtual int put(IDBMeta* dbMeta);
/**
* serialize to string
*/
virtual int toString(std::string& s);
/**
* parse from readonly ptr
*/
virtual int parse(const void* ptr, size_t size);
/**
* if parsed success
*/
virtual bool parsedOK();
virtual size_t getRealSize();
virtual int eraseMapIterator();
virtual int getFromMapIterator(const char** dbName, IDBMeta** dbMeta);
virtual int nextMapIterator(bool erase);
virtual int resetMapIterator();
virtual int erase()
{
return 0;
}
virtual int erase(const char* dbName, bool delayDeleteMeta);
virtual int erase(const char* dbName, const char* tbName, bool delayDeleteMeta);
public:
virtual void setUserData(void* data);
virtual void* getUserData();
private:
MetaDataCollectionInfo* m_coll;
void* m_userData;
};
} // namespace logmessage
} // namespace oceanbase
#endif
include/MsgHeader.h 0 → 100644
浏览文件 @ bf5d80ed
#ifndef _MSG_HEADER_H_
#define _MSG_HEADER_H_
#include <cstdint>
namespace oceanbase {
namespace logmessage {
enum { MT_UNKNOWN = 0, MT_META, MT_FIXED, MT_VAR, MT_EXT };
struct MsgHeader {
uint16_t m_msgType;
uint16_t m_version;
uint32_t m_size;
};
} // namespace logmessage
} // namespace oceanbase
#endif
include/MsgType.h 0 → 100644
浏览文件 @ bf5d80ed
#ifndef _MSG_TYPE_H_
#define _MSG_TYPE_H_
#include <cstdint>
namespace oceanbase {
namespace logmessage {
enum {
DT_UNKNOWN = 0x0000,
DT_INT8 = 0x0001,
DT_UINT8 = 0x0002,
DT_INT16 = 0x0003,
DT_UINT16 = 0x0004,
DT_INT32 = 0x0005,
DT_UINT32 = 0x0006,
DT_INT64 = 0x0007,
DT_UINT64 = 0x0008,
DT_FLOAT = 0x0009,
DT_DOUBLE = 0x000a,
DT_STRING = 0x000b,
TOTAL_DT,
DT_MASK = 0x000F,
DC_ARRAY = 0x0010,
DC_NULL = 0x0020,
DC_MASK = 0x0030,
};
const char STUFF_CHAR = '\0';
typedef uint8_t DT_TYPE;
typedef uint32_t STRLEN_TYPE;
typedef uint32_t COUNT_TYPE;
typedef uint32_t OFFSET_TYPE;
class MsgType {
public:
static int getValType(const char* typeName);
};
} // namespace logmessage
} // namespace oceanbase
#endif
include/StrArray.h 0 → 100644
浏览文件 @ bf5d80ed
#ifndef _STR_ARRAY_H_
#define _STR_ARRAY_H_
#include <sys/types.h>
namespace oceanbase {
namespace logmessage {
class StrArray {
protected:
StrArray() = default;
public:
virtual ~StrArray() = default;
virtual size_t size() = 0;
virtual int elementAt(int i, const char*& s, size_t& length) = 0;
virtual const char* operator[](int i) = 0;
};
} // namespace logmessage
} // namespace oceanbase
#endif
include/UserData.h 0 → 100644
浏览文件 @ bf5d80ed
#ifndef __USER_DATA_H__
#define __USER_DATA_H__
namespace oceanbase {
namespace logmessage {
class UserDataInterface {
public:
virtual ~UserDataInterface()
{}
virtual void setUserData(void* data) = 0;
virtual void* getUserData() = 0;
};
} // namespace logmessage
} // namespace oceanbase
#endif
src/CMakeLists.txt 0 → 100644
浏览文件 @ bf5d80ed
set(SRC_LIST
MsgType.cpp
LogMsgBuf.cpp
MsgVarArea.cpp
LogRecord.cpp
MetaInfo.cpp
LogMsgFactory.cpp
)
add_library(oblogmsg SHARED ${SRC_LIST})
add_library(oblogmsg_static STATIC ${SRC_LIST})
target_link_libraries(oblogmsg PUBLIC oblogmsg_base)
target_link_libraries(oblogmsg_static PUBLIC oblogmsg_base)
set_target_properties(oblogmsg_static PROPERTIES OUTPUT_NAME "oblogmsg")
src/LogMsgBuf.cpp 0 → 100644
浏览文件 @ bf5d80ed
#include <cstdint>
#include <cstring>
#include "MsgType.h"
#include "LogMsgBuf.h"
#include "BinLogBuf.h"
#include "MsgHeader.h"
#include "Endian.h"
namespace oceanbase {
namespace logmessage {
typedef struct _LogMsgBuf {
char* buf;
size_t bufSize;
size_t bufPos;
char* defaultBuf;
size_t avg_size;
} LogMsgBuf;
#define LogMsgHeadSize (sizeof(MsgHeader) + sizeof(COUNT_TYPE))
#define DefaultLogMsgBufSize 1024 * 1024 * 32
#define LOG_MSG_MAX_TAIL_SIZE 1024
static LogMsgBuf lmb_global = {NULL, 0, 0, NULL, 0};
static __thread LogMsgBuf* lmb = &lmb_global;
static inline void lm_check_buf(LogMsgBuf* lmbuf, size_t size, char*& pos, STRLEN_TYPE*& s, char*& head)
{
if (lmbuf->bufSize < size + (pos - lmbuf->buf) + LOG_MSG_MAX_TAIL_SIZE) {
char* tmp = new char[lmbuf->bufSize = (pos - lmbuf->buf) + 2 * (size) + LOG_MSG_MAX_TAIL_SIZE];
memcpy(tmp, lmbuf->buf, pos - lmbuf->buf);
pos = tmp + (pos - lmbuf->buf);
s = (STRLEN_TYPE*)(tmp + ((char*)s - lmbuf->buf));
head = tmp + ((char*)(head)-lmbuf->buf);
if (lmbuf->buf != lmbuf->defaultBuf)
delete[] lmbuf->buf;
lmbuf->buf = tmp;
}
}
int LogMsgLocalInit()
{
LogMsgLocalDestroy();
if (NULL == (lmb = new LogMsgBuf))
return -1;
if (NULL == (lmb->buf = new char[lmb->bufSize = DefaultLogMsgBufSize])) {
delete lmb;
lmb = &lmb_global;
return -1;
}
lmb->defaultBuf = lmb->buf;
lmb->bufPos = 0;
return 0;
}
void LogMsgLocalDestroy()
{
if (lmb == &lmb_global)
return;
if (lmb != NULL) {
if (lmb->buf != NULL)
delete[] lmb->buf;
if (lmb->defaultBuf != lmb->buf && lmb->defaultBuf != NULL)
delete[] lmb->defaultBuf;
delete lmb;
lmb = &lmb_global;
}
}
int LogMsgInit()
{
if (lmb != &lmb_global) {
LogMsgLocalDestroy();
lmb = &lmb_global;
}
LogMsgDestroy();
if (NULL == (lmb_global.buf = new char[lmb_global.bufSize = DefaultLogMsgBufSize]))
return -1;
lmb_global.defaultBuf = lmb_global.buf;
lmb_global.bufPos = 0;
return 0;
}
void LogMsgDestroy()
{
if (lmb != &lmb_global)
return;
if (lmb->buf != NULL)
delete[] lmb->buf;
if (lmb->defaultBuf != lmb->buf && lmb->defaultBuf != NULL)
delete[] lmb->defaultBuf;
lmb->buf = lmb->defaultBuf = NULL;
}
const char* LogMsgGetValueByOffset(size_t offset)
{
return lmb->buf + offset + LogMsgHeadSize + sizeof(DT_TYPE) + sizeof(COUNT_TYPE);
}
size_t LogMsgAppendString(const char* string, size_t size)
{
size_t offset = lmb->bufPos - LogMsgHeadSize;
*(DT_TYPE*)(lmb->buf + lmb->bufPos) = (DT_TYPE)(DT_STRING);
if (string == NULL) {
(*(DT_TYPE*)(lmb->buf + lmb->bufPos)) |= DC_NULL;
return (lmb->bufPos += sizeof(DT_TYPE));
}
*(STRLEN_TYPE*)(lmb->buf + lmb->bufPos + sizeof(DT_TYPE)) = size + 1;
toLeEndian(lmb->buf + lmb->bufPos + sizeof(DT_TYPE), sizeof(STRLEN_TYPE));
memcpy(lmb->buf + lmb->bufPos + sizeof(DT_TYPE) + sizeof(STRLEN_TYPE), string, size);
lmb->bufPos += (sizeof(DT_TYPE) + sizeof(STRLEN_TYPE) + size) + 1;
lmb->buf[lmb->bufPos - 1] = 0;
return offset;
}
size_t LogMsgAppendString(const std::string& string)
{
size_t offset = lmb->bufPos - LogMsgHeadSize, size = string.size();
*(DT_TYPE*)(lmb->buf + lmb->bufPos) = (DT_TYPE)(DT_STRING);
if (size == 0) {
(*(DT_TYPE*)(lmb->buf + lmb->bufPos)) |= DC_NULL;
return (lmb->bufPos += sizeof(DT_TYPE));
}
*(STRLEN_TYPE*)(lmb->buf + lmb->bufPos + sizeof(DT_TYPE)) = size + 1;
toLeEndian(lmb->buf + lmb->bufPos + sizeof(DT_TYPE), sizeof(STRLEN_TYPE));
strncpy(lmb->buf + lmb->bufPos + sizeof(DT_TYPE) + sizeof(STRLEN_TYPE), string.c_str(), size + 1);
lmb->bufPos += (sizeof(DT_TYPE) + sizeof(STRLEN_TYPE) + size + 1);
return offset;
}
size_t LogMsgAppendBuf(const char* data, size_t size)
{
size_t offset = lmb->bufPos - LogMsgHeadSize;
memcpy(lmb->buf + lmb->bufPos, data, size);
lmb->bufPos += size;
return offset;
}
size_t LogMsgAppendBuf(const std::string& string)
{
size_t offset = lmb->bufPos - LogMsgHeadSize;
memcpy(lmb->buf + lmb->bufPos, string.c_str(), string.size());
lmb->bufPos += string.size();
return offset;
}
size_t LogMsgAppendBuf(const BinLogBuf* sa, size_t size)
{
size_t offset = lmb->bufPos - LogMsgHeadSize;
*(DT_TYPE*)(lmb->buf + lmb->bufPos) = (DT_TYPE)(DT_STRING | DC_ARRAY);
lmb->bufPos += sizeof(DT_TYPE);
if (sa == NULL || size == 0) {
*(COUNT_TYPE*)(lmb->buf + lmb->bufPos) = (COUNT_TYPE)(0);
toLeEndian(lmb->buf + lmb->bufPos, sizeof(COUNT_TYPE));
lmb->bufPos += sizeof(COUNT_TYPE);
return offset;
}
*(COUNT_TYPE*)(lmb->buf + lmb->bufPos) = (COUNT_TYPE)(size);
toLeEndian(lmb->buf + lmb->bufPos, sizeof(COUNT_TYPE));
lmb->bufPos += sizeof(COUNT_TYPE);
COUNT_TYPE i = 0;
char *pos = lmb->buf + lmb->bufPos, *head;
STRLEN_TYPE* s = (STRLEN_TYPE*)pos;
pos += sizeof(STRLEN_TYPE) * (size + 1);
head = pos;
for (i = 0; i < size; ++i) {
s[i] = pos - head;
toLeEndian(&(s[i]), sizeof(STRLEN_TYPE));
if (sa[i].buf != NULL) {
lm_check_buf(lmb, sa[i].buf_used_size, pos, s, head);
memcpy(pos, sa[i].buf, sa[i].buf_used_size);
pos[sa[i].buf_used_size] = 0;
pos += (sa[i].buf_used_size + 1);
}
}
s[i] = pos - head;
toLeEndian(&(s[i]), sizeof(STRLEN_TYPE));
lmb->bufPos = pos - lmb->buf;
return offset;
}
size_t LogMsgAppendStringArray(std::vector<std::string*>& sa)
{
size_t offset = lmb->bufPos - LogMsgHeadSize, size = sa.size();
*(DT_TYPE*)(lmb->buf + lmb->bufPos) = (DT_TYPE)(DT_STRING | DC_ARRAY);
lmb->bufPos += sizeof(DT_TYPE);
*(COUNT_TYPE*)(lmb->buf + lmb->bufPos) = (COUNT_TYPE)(sa.size());
toLeEndian(lmb->buf + lmb->bufPos, sizeof(COUNT_TYPE));
if ((COUNT_TYPE)(sa.size()) == 0) {
lmb->bufPos += sizeof(COUNT_TYPE);
return offset;
}
lmb->bufPos += sizeof(COUNT_TYPE);
STRLEN_TYPE len;
COUNT_TYPE i = 0;
char *pos = lmb->buf + lmb->bufPos, *head;
STRLEN_TYPE* s = (STRLEN_TYPE*)pos;
pos += sizeof(STRLEN_TYPE) * (size + 1);
head = pos;
for (i = 0; i < size; ++i) {
s[i] = pos - head;
toLeEndian(&(s[i]), sizeof(STRLEN_TYPE));
if (sa[i] != NULL) {
len = sa[i]->size();
lm_check_buf(lmb, len, pos, s, head);
memcpy(pos, sa[i]->c_str(), len);
pos[len] = 0;
pos += (len + 1);
}
}
s[i] = pos - head;
toLeEndian(&(s[i]), sizeof(STRLEN_TYPE));
lmb->bufPos = pos - lmb->buf;
return offset;
}
size_t LogMsgAppendStringArray(const char** sa, size_t size)
{
size_t offset = lmb->bufPos - LogMsgHeadSize;
*(DT_TYPE*)(lmb->buf + lmb->bufPos) = (DT_TYPE)(DT_STRING | DC_ARRAY);
lmb->bufPos += sizeof(DT_TYPE);
*(COUNT_TYPE*)(lmb->buf + lmb->bufPos) = (COUNT_TYPE)(size);
toLeEndian(lmb->buf + lmb->bufPos, sizeof(COUNT_TYPE));
if ((COUNT_TYPE)(size) == 0) {
lmb->bufPos += sizeof(COUNT_TYPE);
return offset;
}
lmb->bufPos += sizeof(COUNT_TYPE);
STRLEN_TYPE len;
COUNT_TYPE i = 0;
char *pos = lmb->buf + lmb->bufPos, *head;
STRLEN_TYPE* s = (STRLEN_TYPE*)pos;
pos += sizeof(STRLEN_TYPE) * (size + 1);
head = pos;
for (i = 0; i < size; ++i) {
s[i] = pos - head;
toLeEndian(&(s[i]), sizeof(STRLEN_TYPE));
if (sa[i] != NULL) {
len = strlen(sa[i]) + 1;
lm_check_buf(lmb, len, pos, s, head);
memcpy(pos, sa[i], len);
pos += len;
}
}
s[i] = pos - head;
toLeEndian(&(s[i]), sizeof(STRLEN_TYPE));
lmb->bufPos = pos - lmb->buf;
return offset;
}
size_t LogMsgAppendDataArray(std::vector<long>& sa)
{
size_t offset = lmb->bufPos - LogMsgHeadSize;
*(DT_TYPE*)(lmb->buf + lmb->bufPos) = (DT_TYPE)(DT_INT64 | DC_ARRAY);
lmb->bufPos += sizeof(DT_TYPE);
*(COUNT_TYPE*)(lmb->buf + lmb->bufPos) = (COUNT_TYPE)(sa.size());
toLeEndian(lmb->buf + lmb->bufPos, sizeof(COUNT_TYPE));
if ((COUNT_TYPE)(sa.size()) == 0) {
lmb->bufPos += sizeof(COUNT_TYPE);
return offset;
}
lmb->bufPos += sizeof(COUNT_TYPE);
COUNT_TYPE i = 0, j = sa.size();
char* pos = lmb->buf + lmb->bufPos;
pos += sizeof(STRLEN_TYPE) * i;
for (i = 0; i < j; ++i) {
*(long*)pos = sa[i];
toLeEndian(pos, sizeof(long));
pos += sizeof(long);
}
lmb->bufPos = pos - lmb->buf;
return offset;
}
size_t LogMsgAppendDataArray(uint8_t* sa, size_t size)
{
size_t offset = lmb->bufPos - LogMsgHeadSize;
*(DT_TYPE*)(lmb->buf + lmb->bufPos) = (DT_TYPE)(DT_UINT8 | DC_ARRAY);
lmb->bufPos += sizeof(DT_TYPE);
*(COUNT_TYPE*)(lmb->buf + lmb->bufPos) = (COUNT_TYPE)(size);
toLeEndian(lmb->buf + lmb->bufPos, sizeof(COUNT_TYPE));
if ((COUNT_TYPE)(size) == 0 || sa == NULL) {
lmb->bufPos += sizeof(COUNT_TYPE);
return offset;
}
lmb->bufPos += sizeof(COUNT_TYPE);
memcpy(lmb->buf + lmb->bufPos, sa, size);
lmb->bufPos += size;
return offset;
}
void LogMsgSetHead(size_t size)
{
lmb->bufPos = size + LogMsgHeadSize + sizeof(DT_TYPE) + sizeof(COUNT_TYPE);
if (lmb->buf != lmb->defaultBuf) {
if (lmb->avg_size > DefaultLogMsgBufSize) {
if (lmb->avg_size < ((lmb->bufSize + DefaultLogMsgBufSize) >> 1)) {
delete[] lmb->buf;
lmb->buf = new char[lmb->bufSize = (lmb->bufSize + DefaultLogMsgBufSize) >> 1];
}
} else {
delete[] lmb->buf;
lmb->buf = lmb->defaultBuf;
lmb->bufSize = DefaultLogMsgBufSize;
}
}
}
void LogMsgCopyHead(const char* head, size_t size)
{
((struct MsgHeader*)lmb->buf)->m_msgType = MT_VAR;
toLeEndian(&(((struct MsgHeader*)lmb->buf)->m_msgType), sizeof(uint16_t));
((struct MsgHeader*)lmb->buf)->m_version = 1;
toLeEndian(&(((struct MsgHeader*)lmb->buf)->m_version), sizeof(uint16_t));
((struct MsgHeader*)lmb->buf)->m_size = lmb->bufPos - (sizeof(struct MsgHeader));
toLeEndian(&(((struct MsgHeader*)lmb->buf)->m_size), sizeof(uint32_t));
*(COUNT_TYPE*)(lmb->buf + sizeof(struct MsgHeader)) = 0;
toLeEndian(lmb->buf + sizeof(struct MsgHeader), sizeof(COUNT_TYPE));
*(lmb->buf + LogMsgHeadSize) = (DT_TYPE)(DT_UINT8 | DC_ARRAY);
toLeEndian(lmb->buf + LogMsgHeadSize, sizeof(DT_TYPE));
*(COUNT_TYPE*)(lmb->buf + LogMsgHeadSize + sizeof(DT_TYPE)) = size;
toLeEndian(lmb->buf + LogMsgHeadSize + sizeof(DT_TYPE), sizeof(COUNT_TYPE));
memcpy(lmb->buf + LogMsgHeadSize + sizeof(DT_TYPE) + sizeof(COUNT_TYPE), head, size);
}
void LogMsgFroceSetHeadSize(size_t size)
{
*(COUNT_TYPE*)(lmb->buf + LogMsgHeadSize + sizeof(DT_TYPE)) = size;
toLeEndian(lmb->buf + LogMsgHeadSize + sizeof(DT_TYPE), sizeof(COUNT_TYPE));
}
const char* LogMsgGetString(size_t* size)
{
*size = lmb->bufPos;
if (lmb->buf != lmb->defaultBuf && lmb->bufPos > DefaultLogMsgBufSize)
lmb->avg_size = ((((long)lmb->avg_size >> 1) + (long)lmb->bufPos) >> 1) + (lmb->bufPos >> 2);
else
lmb->avg_size = (((long)lmb->avg_size + (long)lmb->bufPos) >> 1);
return lmb->buf;
}
} // namespace logmessage
} // namespace oceanbase
src/LogMsgFactory.cpp 0 → 100644
浏览文件 @ bf5d80ed
#include "LogMsgFactory.h"
#include "MetaInfo.h"
#include "LogRecord.h"
using namespace std;
namespace oceanbase {
namespace logmessage {
string LogMsgFactory::DFT_ColMeta = "ColMetaImpl";
string LogMsgFactory::DFT_TableMeta = "TableMetaImpl";
string LogMsgFactory::DFT_DBMeta = "DBMetaImpl";
string LogMsgFactory::DFT_METAS = "MetaDataCollectionsImpl";
string LogMsgFactory::DFT_LR = "LogRecordImpl";
IColMeta* LogMsgFactory::createColMeta(const string& type)
{
if (type == DFT_ColMeta)
return new IColMeta();
else
return NULL;
}
IColMeta* LogMsgFactory::createColMeta(const string& type, const char* ptr, size_t size)
{
if (type == DFT_ColMeta)
return new IColMeta(ptr, size);
else
return NULL;
}
ITableMeta* LogMsgFactory::createTableMeta(const string& type)
{
if (type == DFT_TableMeta)
return new ITableMeta();
else
return NULL;
}
ITableMeta* LogMsgFactory::createTableMeta(const string& type, const char* ptr, size_t size)
{
if (type == DFT_TableMeta)
return new ITableMeta(ptr, size);
else
return NULL;
}
IDBMeta* LogMsgFactory::createDBMeta(const string& type)
{
if (type == DFT_DBMeta)
return new IDBMeta();
else
return NULL;
}
IDBMeta* LogMsgFactory::createDBMeta(const string& type, const char* ptr, size_t size)
{
if (type == DFT_DBMeta)
return new IDBMeta(ptr, size);
else
return NULL;
}
IMetaDataCollections* LogMsgFactory::createMetaDataCollections(const string& type)
{
if (type == DFT_METAS)
return new IMetaDataCollections();
return NULL;
}
IMetaDataCollections* LogMsgFactory::createMetaDataCollections(
const string& type, const char* ptr, size_t size, bool clearGarbage)
{
if (type == DFT_METAS)
return new IMetaDataCollections(ptr, size, clearGarbage);
return NULL;
}
ILogRecord* LogMsgFactory::createLogRecord(const std::string& type, bool creating)
{
if (type == DFT_LR)
return new LogRecordImpl(creating, true);
return NULL;
}
ILogRecord* LogMsgFactory::createLogRecord(const std::string& type, const char* ptr, size_t size)
{
if (type == DFT_LR)
return new LogRecordImpl(ptr, size);
return NULL;
}
void LogMsgFactory::destroy(ILogRecord*& record)
{
if (record) {
record->clear();
delete record;
record = NULL;
}
}
void LogMsgFactory::destroyWithUserMemory(ILogRecord*& record)
{
if (record) {
delete record;
record = NULL;
}
}
void LogMsgFactory::destroy(IColMeta*& colMeta)
{
if (colMeta) {
delete colMeta;
colMeta = NULL;
}
}
void LogMsgFactory::destroy(ITableMeta*& tableMeta)
{
if (tableMeta) {
delete tableMeta;
tableMeta = NULL;
}
}
void LogMsgFactory::destroy(IDBMeta*& dbMeta)
{
if (dbMeta) {
delete dbMeta;
dbMeta = NULL;
}
}
void LogMsgFactory::destroy(IMetaDataCollections*& metaColls)
{
if (metaColls) {
delete metaColls;
metaColls = NULL;
}
}
} // namespace logmessage
} // namespace oceanbase
src/LogRecord.cpp 0 → 100644
浏览文件 @ bf5d80ed
此差异已折叠。
src/MetaInfo.cpp 0 → 100644
浏览文件 @ bf5d80ed
此差异已折叠。
src/MsgType.cpp 0 → 100644
浏览文件 @ bf5d80ed
#include "MsgType.h"
#include <typeinfo>
#include <cstring>
namespace oceanbase {
namespace logmessage {
// typename => type
static struct {
const char* tn;
int vt;
} g_nameTypeMap[] = {{typeid(bool).name(), DT_INT8},
{typeid(char).name(), DT_INT8},
{typeid(int8_t).name(), DT_INT8},
{typeid(uint8_t).name(), DT_UINT8},
{typeid(int16_t).name(), DT_INT16},
{typeid(uint16_t).name(), DT_UINT16},
{typeid(int32_t).name(), DT_INT32},
{typeid(uint32_t).name(), DT_UINT32},
{typeid(int64_t).name(), DT_INT64},
{typeid(uint64_t).name(), DT_UINT64},
{typeid(long).name(), DT_INT64},
{typeid(unsigned long).name(), DT_UINT64},
{typeid(float).name(), DT_FLOAT},
{typeid(double).name(), DT_DOUBLE},
{0}};
int MsgType::getValType(const char* typeName)
{
for (int i = 0; g_nameTypeMap[i].tn != NULL; ++i) {
if (strcmp(g_nameTypeMap[i].tn, typeName) == 0) {
return g_nameTypeMap[i].vt;
}
}
return DT_UNKNOWN;
}
} // namespace logmessage
} // namespace oceanbase
src/MsgVarArea.cpp 0 → 100644
浏览文件 @ bf5d80ed
#include "MsgVarArea.h"
#include "MsgType.h"
#include "Endian.h"
#include <string>
#include <cstring>
#include <cstdint>
namespace oceanbase {
namespace logmessage {
#define VAR_MSG_VERSION 1
/*
* V1 format:
* <field-count>
* <type of field><value of field>
* <type of field><value of field>
* ...
* <type of field><value of field>
*/
// ------------ VarMsgArea -------------
class MsgStrArray : public StrArray {
public:
MsgStrArray(const char*& saPtr, size_t& count, const OFFSET_TYPE*& offsets);
virtual ~MsgStrArray();
virtual size_t size();
virtual int elementAt(int i, const char*& s, size_t& length);
virtual const char* operator[](int i);
private:
const char* m_saPtr;
size_t m_count;
const OFFSET_TYPE* m_offsets;
};
template size_t MsgVarArea::appendStringArray<std::vector<std::string*>>(std::vector<std::string*>&);
template size_t MsgVarArea::appendStringArray<std::vector<std::string>>(std::vector<std::string>&);
template size_t MsgVarArea::appendStringArray<std::vector<const char*>>(std::vector<const char*>&);
// -------- MsgVarArea -----------
MsgVarArea::MsgVarArea(bool creating)
: m_count(0),
m_ptr(NULL),
m_size(0),
m_creating(creating),
m_parsedOK(false),
m_areaHeader(NULL),
m_areaPtr(NULL),
m_areaSize(0),
m_areaEnd(NULL)
{
if (creating)
m_data.append(sizeof(VarAreaHeader), STUFF_CHAR); // reserved
}
MsgVarArea::MsgVarArea(const void* ptr, size_t size)
: m_ptr(NULL),
m_size(0),
m_creating(false),
m_parsedOK(false),
m_areaHeader(NULL),
m_areaPtr(NULL),
m_areaSize(0),
m_areaEnd(NULL)
{
parse(ptr, size);
}
size_t MsgVarArea::append(const char* typeName, const void* ptr, size_t size)
{
// type、val
DT_TYPE valType = (DT_TYPE)MsgType::getValType(typeName);
if (valType == DT_UNKNOWN)
return (size_t)-1;
size_t offset = m_data.length() - sizeof(VarAreaHeader);
m_data.append((const char*)&valType, sizeof(DT_TYPE));
m_data.append((const char*)ptr, size);
m_count++;
afterAppending();
return offset;
}
size_t MsgVarArea::appendData(const std::string& s)
{
size_t offset = m_data.length() - sizeof(VarAreaHeader);
m_data.append(s);
m_count++;
afterAppending();
return offset;
}
size_t MsgVarArea::appendString(const char* s, size_t length)
{
// type、len、val
DT_TYPE valType = DT_STRING;
size_t offset = m_data.length() - sizeof(VarAreaHeader);
if (s == NULL) {
valType |= DC_NULL;
m_data.append((const char*)&valType, sizeof(DT_TYPE));
} else {
STRLEN_TYPE len = (STRLEN_TYPE)(length);
STRLEN_TYPE lenEndia = len;
toLeEndian(&lenEndia, sizeof(STRLEN_TYPE));
m_data.append((const char*)&valType, sizeof(DT_TYPE));
m_data.append((const char*)&lenEndia, sizeof(STRLEN_TYPE));
m_data.append((const char*)s, len);
}
m_count++;
afterAppending();
return offset;
}
size_t MsgVarArea::appendString(const char* s)
{
return appendString(s, s == NULL ? 0 : strlen(s) + 1);
}
size_t MsgVarArea::appendString(std::string* s)
{
return appendString(s == NULL ? NULL : s->c_str(), s == NULL ? 0 : s->length() + 1);
}
std::string MsgVarArea::createStringArrayData(
const char** sa, COUNT_TYPE count, STRLEN_TYPE* slen, bool appendTail, bool appendAlway)
{
// type、count、strOffsets、strings
std::string data;
DT_TYPE valType = (DT_TYPE)(DT_STRING | DC_ARRAY);
data.append((const char*)&valType, sizeof(DT_TYPE));
if (sa == NULL || count == 0) {
data.append(sizeof(COUNT_TYPE), '\0'); // count == 0
return data;
}
COUNT_TYPE countEndia = count;
toLeEndian(&countEndia, sizeof(COUNT_TYPE));
data.append((const char*)&countEndia, sizeof(COUNT_TYPE));
OFFSET_TYPE* strOffsets = new OFFSET_TYPE[count + 1];
OFFSET_TYPE* strOffsetsEndia = new OFFSET_TYPE[count + 1];
OFFSET_TYPE strOffset = 0;
// for strOffsets
STRLEN_TYPE len;
COUNT_TYPE i, j;
for (i = 0; i < count; ++i) {
strOffsets[i] = strOffset;
strOffsetsEndia[i] = strOffset;
toLeEndian(&(strOffsetsEndia[i]), sizeof(OFFSET_TYPE));
strOffset += slen[i];
}
strOffsets[i] = strOffset;
strOffsetsEndia[i] = strOffset;
toLeEndian(&(strOffsetsEndia[i]), sizeof(OFFSET_TYPE));
data.append((const char*)strOffsetsEndia, sizeof(OFFSET_TYPE) * (count + 1));
for (i = 0, j = 1; i < count; ++i, ++j) {
len = strOffsets[j] - strOffsets[i];
if (len > 0) {
if (appendTail) {
data.append(sa[i], len - 1);
data.append("\0", 1);
} else
data.append(sa[i], len);
} else if (appendAlway)
data.append(sa[i], len);
}
delete[] strOffsets;
delete[] strOffsetsEndia;
return data;
}
size_t MsgVarArea::appendStringArray(const BinLogBuf* sa, size_t size)
{
// type、count、strOffsets、strings
COUNT_TYPE i, count = (COUNT_TYPE)size;
STRLEN_TYPE* slen = new STRLEN_TYPE[count];
const char* strData[count];
size_t offset = m_data.length() - sizeof(VarAreaHeader);
for (i = 0; i < count; ++i) {
if (sa[i].buf == NULL)
slen[i] = 0;
else
slen[i] = (STRLEN_TYPE)(sa[i].buf_used_size + 1);
if (slen[i] > 0)
strData[i] = sa[i].buf;
}
m_data.append(createStringArrayData(strData, count, slen, true, false));
if (sa != NULL && size != 0)
m_count++;
afterAppending();
delete[] slen;
return offset;
}
size_t MsgVarArea::appendStringArray(const char** sa, size_t size)
{
size_t offset = m_data.length() - sizeof(VarAreaHeader);
m_data.append(createStringArrayData(sa, size));
if (sa != NULL && size != 0)
m_count++;
afterAppending();
return offset;
}
std::string MsgVarArea::createStringArrayData(const char** sa, size_t size)
{
// type、count、strOffsets、strings
std::string data;
COUNT_TYPE i, count = (COUNT_TYPE)size;
STRLEN_TYPE* slen = new STRLEN_TYPE[count];
for (i = 0; i < count; ++i) {
if (sa[i] == NULL)
slen[i] = 0;
else
slen[i] = (STRLEN_TYPE)(strlen(sa[i]) + 1);
}
data = createStringArrayData(sa, count, slen, false, false);
delete[] slen;
return data;
}
std::string MsgVarArea::createStringArrayData(std::vector<std::string*>& sa)
{
// type、count、strOffsets、strings
std::string data;
COUNT_TYPE i, count = (COUNT_TYPE)sa.size();
STRLEN_TYPE* slen = new STRLEN_TYPE[count];
const char* strData[count];
for (i = 0; i < count; ++i) {
if (sa[i] == NULL)
slen[i] = 0;
else
slen[i] = (STRLEN_TYPE)(sa[i]->length() + 1);
if (slen[i] > 0)
strData[i] = sa[i]->c_str();
}
data = createStringArrayData(strData, count, slen, false, false);
delete[] slen;
return data;
}
// deal type vector<string*> vector<string> vector<const char*>
template <typename T>
size_t MsgVarArea::appendStringArray(T& sa)
{
// type、count、strOffsets、strings
COUNT_TYPE count = (COUNT_TYPE)sa.size();
size_t offset = m_data.length() - sizeof(VarAreaHeader);
m_data.append(createStringArrayData(sa));
if (count != 0)
m_count++;
afterAppending();
return offset;
}
std::string MsgVarArea::createStringArrayData(std::vector<std::string>& sa)
{
// type、count、strOffsets、strings
std::string data;
COUNT_TYPE i, count = (COUNT_TYPE)sa.size();
STRLEN_TYPE* slen = new STRLEN_TYPE[count];
const char* strData[count];
for (i = 0; i < count; ++i) {
slen[i] = (STRLEN_TYPE)(sa[i].length() + 1);
strData[i] = sa[i].c_str();
}
data = createStringArrayData(strData, count, slen, false, true);
delete[] slen;
return data;
}
std::string MsgVarArea::createStringArrayData(std::vector<const char*>& sa)
{
// type、count、strOffsets、strings
std::string data;
COUNT_TYPE i, count = (COUNT_TYPE)sa.size();
STRLEN_TYPE* slen = new STRLEN_TYPE[count];
const char* strData[count];
for (i = 0; i < count; ++i) {
if (sa[i] == NULL)
slen[i] = 0;
else
slen[i] = (STRLEN_TYPE)(strlen(sa[i]) + 1);
strData[i] = sa[i];
}
data = createStringArrayData(strData, count, slen, false, false);
delete[] slen;
return data;
}
size_t MsgVarArea::appendArray(const char* typeName, const void* a, size_t elSize, size_t size)
{
// type、count、array
DT_TYPE valType = MsgType::getValType(typeName);
if (valType == DT_UNKNOWN)
return (size_t)-1;
size_t offset = m_data.length() - sizeof(VarAreaHeader);
m_data.append(createArray(typeName, a, elSize, size));
if (a != NULL && elSize != 0 && size != 0) {
m_count++;
}
afterAppending();
return offset;
}
std::string MsgVarArea::createArray(const char* typeName, const void* a, size_t elSize, size_t size)
{
// type、count、array
std::string data;
size_t i = 0;
DT_TYPE valType = MsgType::getValType(typeName);
if (valType == DT_UNKNOWN)
return std::string(" ");
valType |= DC_ARRAY;
COUNT_TYPE count = (COUNT_TYPE)size;
toLeEndian(&count, sizeof(COUNT_TYPE));
data.append((const char*)&valType, sizeof(DT_TYPE));
if (a == NULL || elSize == 0 || size == 0) {
data.append(sizeof(COUNT_TYPE), '\0'); // count
return data;
}
data.append((const char*)&count, sizeof(COUNT_TYPE));
char* aEndia = new char[elSize * size];
memcpy(aEndia, (const char*)a, elSize * size);
for (; i < size; i++) {
toLeEndian(aEndia + i * elSize, elSize);
}
data.append((const char*)aEndia, elSize * size);
delete[] aEndia;
return data;
}
const std::string& MsgVarArea::getMessage()
{
uint32_t size = m_data.length() - sizeof(MsgHeader);
uint16_t mt_var = MT_VAR;
uint16_t msg_version = VAR_MSG_VERSION;
VarAreaHeader header = {
{toLeEndianByType(mt_var), toLeEndianByType(msg_version), toLeEndianByType(size)}, toLeEndianByType(m_count)};
m_data.replace(0, sizeof(header), (const char*)&header, sizeof(header));
// to support fetching
m_ptr = m_data.c_str();
m_size = size;
m_areaHeader = (VarAreaHeader*)m_ptr;
m_areaPtr = (const char*)(m_areaHeader + 1);
m_areaSize = m_data.length() - sizeof(VarAreaHeader);
m_areaEnd = m_areaPtr + m_areaSize;
m_parsedOK = true;
return m_data;
}
void MsgVarArea::clear()
{
m_ptr = NULL;
m_size = 0;
m_areaHeader = NULL;
m_areaPtr = NULL;
m_areaSize = 0;
m_areaEnd = NULL;
m_count = 0;
m_data.clear();
if (m_data.capacity() > 1024 * 20)
std::string().swap(m_data);
m_data.append(sizeof(VarAreaHeader), STUFF_CHAR);
}
size_t MsgVarArea::getRealSize()
{
if (m_parsedOK)
return m_areaSize + sizeof(VarAreaHeader);
return m_data.length();
}
const void* MsgVarArea::getMsgBuf(size_t& size)
{
size = m_size;
return m_ptr;
}
int MsgVarArea::parse(const void* ptr, size_t size)
{
clear();
if (sizeof(MsgHeader) > size)
return -1;
m_areaHeader = (VarAreaHeader*)ptr;
MsgHeader* msgHeader = (MsgHeader*)&(m_areaHeader->m_msgHeader);
if (toLeEndianByType(msgHeader->m_size) + sizeof(MsgHeader) > size)
return -2;
if (toLeEndianByType(msgHeader->m_msgType) != MT_VAR)
return -3;
if (toLeEndianByType(msgHeader->m_version) > VAR_MSG_VERSION)
return -4;
m_areaPtr = (const char*)(m_areaHeader + 1);
m_areaSize = toLeEndianByType(msgHeader->m_size) - (sizeof(VarAreaHeader) - sizeof(MsgHeader));
m_areaEnd = m_areaPtr + m_areaSize;
m_ptr = (const char*)ptr;
m_size = size;
m_parsedOK = true;
return 0;
}
int MsgVarArea::copy(const void* ptr, size_t size)
{
m_data.assign((const char*)ptr, size);
return 0;
}
int MsgVarArea::getField(size_t offset, const void*& ptr, size_t& size)
{
if (!m_parsedOK && !m_creating)
return -1;
if (offset < 0 || offset >= m_areaSize)
return -2;
const char* p = m_areaPtr + offset;
DT_TYPE* t = (DT_TYPE*)p;
if ((*t & DC_ARRAY) != 0) {
if ((*t & DT_MASK) == DT_STRING) {
const char* saPtr;
size_t count;
const OFFSET_TYPE* offsets;
int ret = getStringArray(offset, saPtr, count, offsets);
if (ret != 0)
return ret;
ptr = saPtr;
size = toLeEndianByType(offsets[count]);
return 0;
}
size_t elSize, count;
const void* a;
int ret = getArray(offset, a, elSize, count);
if (ret != 0)
return ret;
ptr = a;
size = elSize * count;
return 0;
}
int vt = *t & DT_MASK;
p += sizeof(DT_TYPE);
ptr = p;
switch (vt) {
case DT_INT8:
case DT_UINT8:
size = sizeof(int8_t);
break;
case DT_INT16:
case DT_UINT16:
size = sizeof(int16_t);
break;
case DT_INT32:
case DT_UINT32:
size = sizeof(int32_t);
break;
case DT_INT64:
case DT_UINT64:
size = sizeof(int64_t);
break;
case DT_FLOAT:
size = sizeof(float);
break;
case DT_DOUBLE:
size = sizeof(double);
break;
case DT_STRING:
if ((*t & DC_NULL) != 0) {
size = 0;
ptr = NULL;
} else {
STRLEN_TYPE len = *(STRLEN_TYPE*)p;
toLeEndian(&len, sizeof(STRLEN_TYPE));
size = len;
ptr = p + sizeof(STRLEN_TYPE);
}
break;
default:
return -3;
}
if (p + size > m_areaEnd)
return -4;
return 0;
}
int MsgVarArea::getString(size_t offset, const char*& s, size_t& length)
{
if (!m_parsedOK && !m_creating)
return -1;
if (offset < 0 || offset >= m_areaSize)
return -2;
const char* p = m_areaPtr + offset;
DT_TYPE* t = (DT_TYPE*)p;
if ((*t & DC_ARRAY) != 0)
return -3;
if ((*t & DC_NULL) != 0) {
s = NULL;
length = 0;
return 0;
}
p += sizeof(DT_TYPE);
if (p + sizeof(STRLEN_TYPE) >= m_areaEnd)
return -4;
STRLEN_TYPE len = *(STRLEN_TYPE*)p;
toLeEndian(&len, sizeof(STRLEN_TYPE));
p += sizeof(STRLEN_TYPE);
if (p + len > m_areaEnd)
return -5;
s = p;
length = len - 1;
return 0;
}
StrArray* MsgVarArea::getStringArray(size_t offset)
{
const char* saPtr;
size_t count;
const OFFSET_TYPE* offsets;
int ret = getStringArray(offset, saPtr, count, offsets);
if (ret != 0)
return NULL;
if (count == 0)
return NULL;
return new MsgStrArray(saPtr, count, offsets);
}
int MsgVarArea::getStringArray(size_t offset, const char*& saPtr, size_t& count, const OFFSET_TYPE*& offsets)
{
if (!m_parsedOK && !m_creating)
return -1;
if (offset < 0 || offset >= m_areaSize)
return -2;
const char* p = m_areaPtr + offset;
DT_TYPE* t = (DT_TYPE*)p;
if ((*t & DC_ARRAY) == 0 || (*t & DT_MASK) != DT_STRING)
return -3;
p += sizeof(DT_TYPE);
if (p + sizeof(COUNT_TYPE) > m_areaEnd)
return -4;
COUNT_TYPE Count = *(COUNT_TYPE*)p;
toLeEndian(&Count, sizeof(COUNT_TYPE));
count = Count;
if (count == 0) {
saPtr = NULL;
offsets = NULL;
return 0;
}
p += sizeof(COUNT_TYPE);
if (p + (count + 1) * sizeof(OFFSET_TYPE) > m_areaEnd)
return -5;
OFFSET_TYPE* strOffsets = (OFFSET_TYPE*)p;
OFFSET_TYPE strOffset;
p += (count + 1) * sizeof(OFFSET_TYPE);
strOffset = strOffsets[count];
toLeEndian(&strOffset, sizeof(OFFSET_TYPE));
if (p + strOffset > m_areaEnd)
return -6;
saPtr = p;
offsets = strOffsets;
return 0;
}
void MsgVarArea::getString(size_t offset, const int off, char*& v, size_t& size)
{
const char* pos = m_areaPtr + offset;
COUNT_TYPE count = *(COUNT_TYPE*)(pos + sizeof(DT_TYPE));
toLeEndian(&count, sizeof(COUNT_TYPE));
if (offset < 0 || offset >= m_areaSize || off < 0 || off > count) {
v = NULL;
size = 0;
return;
}
OFFSET_TYPE* ot = (OFFSET_TYPE*)(pos + sizeof(DT_TYPE) + sizeof(COUNT_TYPE));
if (toLeEndianByType(ot[off + 1]) <= toLeEndianByType(ot[off])) {
v = NULL;
size = 0;
return;
}
v = (char*)ot + sizeof(OFFSET_TYPE) * (1 + count + toLeEndianByType(ot[off]));
size = toLeEndianByType(ot[off + 1]) - toLeEndianByType(ot[off]) - 1;
}
int MsgVarArea::getBuf(size_t offset, const void*& a, size_t& size)
{
const char* p = m_areaPtr + offset;
COUNT_TYPE count = *(COUNT_TYPE*)(p + sizeof(DT_TYPE));
toLeEndian(&count, sizeof(COUNT_TYPE));
size = count;
if (p + size > m_areaEnd)
return -2;
a = p + sizeof(DT_TYPE) + sizeof(COUNT_TYPE);
return 0;
}
void MsgVarArea::getStringArrayData(uint32_t offset, const char*& v, size_t& size)
{
v = m_areaPtr + offset;
COUNT_TYPE cnt = *(COUNT_TYPE*)(v + sizeof(DT_TYPE));
toLeEndian(&cnt, sizeof(COUNT_TYPE));
if (cnt == 0) {
v = NULL;
size = 0;
return;
}
OFFSET_TYPE* ot = (OFFSET_TYPE*)(v + sizeof(DT_TYPE) + sizeof(COUNT_TYPE));
size = toLeEndianByType(ot[cnt]) + (cnt + 1) * sizeof(OFFSET_TYPE) + sizeof(DT_TYPE) + sizeof(COUNT_TYPE);
}
int MsgVarArea::getArray(size_t offset, const void*& a, size_t& elSize, size_t& size)
{
if (!m_parsedOK && !m_creating)
return -1;
if (offset < 0 || offset >= m_areaSize)
return -2;
const char* p = m_areaPtr + offset;
DT_TYPE* t = (DT_TYPE*)p;
if ((*t & DC_ARRAY) == 0)
return -3;
p += sizeof(DT_TYPE);
if (p + sizeof(COUNT_TYPE) > m_areaEnd)
return -4;
COUNT_TYPE count = *(COUNT_TYPE*)(p);
toLeEndian(&count, sizeof(COUNT_TYPE));
size = count;
if (size == 0) {
a = NULL;
elSize = 0;
return 0;
}
switch (*t & DT_MASK) {
case DT_INT8:
case DT_UINT8:
elSize = sizeof(int8_t);
break;
case DT_INT16:
case DT_UINT16:
elSize = sizeof(int16_t);
break;
case DT_INT32:
case DT_UINT32:
elSize = sizeof(int32_t);
break;
case DT_INT64:
case DT_UINT64:
elSize = sizeof(int64_t);
break;
case DT_FLOAT:
elSize = sizeof(float);
break;
case DT_DOUBLE:
elSize = sizeof(double);
break;
default:
return -5;
}
p += sizeof(COUNT_TYPE);
if (p + elSize * size > m_areaEnd)
return -6;
a = p;
return 0;
}
void MsgVarArea::afterAppending()
{
m_areaPtr = m_data.c_str() + sizeof(VarAreaHeader);
m_areaSize = m_data.length() - sizeof(VarAreaHeader);
m_areaEnd = m_areaPtr + m_areaSize;
}
const char* MsgVarArea::getString(size_t offset)
{
const char* s;
size_t length;
int ret = getString(offset, s, length);
if (ret == 0)
return s;
return NULL;
}
// cz-string array wrapper
MsgStrArray::MsgStrArray(const char*& saPtr, size_t& count, const uint32_t*& offsets)
: m_saPtr(saPtr), m_count(count), m_offsets(offsets)
{}
MsgStrArray::~MsgStrArray()
{}
size_t MsgStrArray::size()
{
return m_count;
}
int MsgStrArray::elementAt(int i, const char*& s, size_t& length)
{
if (i < 0 || (size_t)i >= m_count)
return -1;
length = toLeEndianByType(m_offsets[i + 1]) - toLeEndianByType(m_offsets[i]);
s = (length > 0) ? m_saPtr + toLeEndianByType(m_offsets[i]) : NULL;
return 0;
}
const char* MsgStrArray::operator[](int i)
{
if (i < 0 || (size_t)i >= m_count)
return NULL;
size_t length = toLeEndianByType(m_offsets[i + 1]) - toLeEndianByType(m_offsets[i]);
return (length > 0) ? m_saPtr + toLeEndianByType(m_offsets[i]) : NULL;
}
} // namespace logmessage
} // namespace oceanbase
src/MsgVarArea.h 0 → 100644
浏览文件 @ bf5d80ed
#ifndef _MSG_VAR_AREA_H_
#define _MSG_VAR_AREA_H_
#include "StrArray.h"
#include "MsgType.h"
#include "MsgHeader.h"
#include <cstdio>
#include <string>
#include <vector>
#include <typeinfo>
#include "BinLogBuf.h"
namespace oceanbase {
namespace logmessage {
// ---------- Header ------------
struct VarAreaHeader {
MsgHeader m_msgHeader;
COUNT_TYPE m_count;
};
class MsgVarArea {
public:
MsgVarArea(bool creating = true);
MsgVarArea(const void* ptr, size_t size);
~MsgVarArea()
{}
// methods for creating
template <typename T>
size_t append(T val)
{
return append(typeid(T).name(), (const void*)&val, sizeof(T));
}
template <typename T>
size_t appendStringArray(T& sa);
size_t appendString(const char* s);
size_t appendData(const std::string& s);
static std::string createStringArrayData(
const char** sa, COUNT_TYPE count, STRLEN_TYPE* slen, bool appendTail, bool appendAlway);
size_t appendString(const char* s, size_t length);
size_t appendString(std::string* s);
size_t appendStringArray(const BinLogBuf* sa, size_t size);
size_t appendStringArray(const char** sa, size_t size);
static std::string createStringArrayData(const char** sa, size_t size);
static std::string createStringArrayData(std::vector<std::string*>& sa);
static std::string createStringArrayData(std::vector<std::string>& sa);
static std::string createStringArrayData(std::vector<const char*>& sa);
template <typename T>
size_t appendArray(T* a, size_t size)
{
return appendArray(typeid(T).name(), (const void*)a, sizeof(T), size);
}
template <typename T>
static std::string createArrayData(T* a, size_t size)
{
return createArray(typeid(T).name(), (const void*)a, sizeof(T), size);
}
const std::string& getMessage();
private:
size_t append(const char* typeName, const void* ptr, size_t size); // return offset
size_t appendArray(const char* typeName, const void* a, size_t elSize, size_t count);
static std::string createArray(const char* typeName, const void* a, size_t elSize, size_t count);
void afterAppending();
public:
// methods for fetching
void clear();
size_t getRealSize();
const void* getMsgBuf(size_t& size);
int parse(const void* ptr, size_t size);
int copy(const void* ptr, size_t size);
int getField(size_t offset, const void*& ptr, size_t& size);
int getString(size_t offset, const char*& s, size_t& length);
const char* getString(size_t offset);
void getString(size_t offset, const int off, char*& v, size_t& size);
StrArray* getStringArray(size_t offset);
int getStringArray(size_t offset, const char*& saPtr, size_t& count, const OFFSET_TYPE*& offsets);
int getArray(size_t offset, const void*& a, size_t& elSize, size_t& size);
int getBuf(size_t offset, const void*& a, size_t& size);
void getStringArrayData(unsigned int offset, const char*& v, size_t& size);
private:
COUNT_TYPE m_count;
std::string m_data; // save header data
const char* m_ptr;
size_t m_size;
bool m_creating;
bool m_parsedOK;
VarAreaHeader* m_areaHeader;
const char* m_areaPtr; // data begin
size_t m_areaSize; // data size
const char* m_areaEnd; // data end
};
} // namespace logmessage
} // namespace oceanbase
#endif
unittest/CMakeLists.txt 0 → 100644
浏览文件 @ bf5d80ed
enable_testing()
add_executable(lr_unittest lrTest.cpp)
add_executable(varArea_unittest varAreaTest.cpp)
add_executable(md_unittest mdTest.cpp)
add_executable(fa_unittest faTest.cpp)
add_executable(memoryleak_unittest memoryleakTest.cpp)
target_link_libraries(lr_unittest PUBLIC oblogmsg gtest)
target_link_libraries(varArea_unittest PUBLIC oblogmsg gtest)
target_include_directories(varArea_unittest PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/../src)
target_link_libraries(md_unittest PUBLIC oblogmsg gtest)
target_link_libraries(fa_unittest PUBLIC oblogmsg gtest)
target_link_libraries(memoryleak_unittest PUBLIC oblogmsg gtest)
add_test(lr_unittest lr_unittest)
add_test(md_unittest md_unittest)
add_test(fa_unittest fa_unittest)
add_test(varArea_unittest varArea_unittest)
add_test(memoryleak_unittest memoryleak_unittest)
unittest/faTest.cpp 0 → 100644
浏览文件 @ bf5d80ed
#include "MetaInfo.h"
#include "LogRecord.h"
#include "LogMsgFactory.h"
#include <gtest/gtest.h>
using namespace oceanbase::logmessage;
TEST(LogMsgFactory, LogMsgFactoryAPI)
{
IColMeta* colMeta = LogMsgFactory::createColMeta();
ASSERT_NE((void*)NULL, (void*)colMeta);
ITableMeta* tableMeta = LogMsgFactory::createTableMeta();
ASSERT_NE((void*)NULL, (void*)tableMeta);
IDBMeta* dbMeta = LogMsgFactory::createDBMeta();
ASSERT_NE((void*)NULL, (void*)dbMeta);
IMetaDataCollections* meta = LogMsgFactory::createMetaDataCollections();
ASSERT_NE((void*)NULL, (void*)meta);
ILogRecord* record = LogMsgFactory::createLogRecord();
ASSERT_NE((void*)NULL, (void*)record);
LogMsgFactory::destroy(colMeta);
ASSERT_EQ((void*)NULL, (void*)colMeta);
LogMsgFactory::destroy(tableMeta);
ASSERT_EQ((void*)NULL, (void*)tableMeta);
LogMsgFactory::destroy(dbMeta);
ASSERT_EQ((void*)NULL, (void*)dbMeta);
LogMsgFactory::destroy(meta);
ASSERT_EQ((void*)NULL, (void*)meta);
LogMsgFactory::destroy(record);
ASSERT_EQ((void*)NULL, (void*)record);
LogMsgFactory::destroy(record);
ASSERT_EQ((void*)NULL, (void*)record);
}
int main(int argc, char* argv[])
{
testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}
unittest/lrTest.cpp 0 → 100644
浏览文件 @ bf5d80ed
此差异已折叠。
unittest/mdTest.cpp 0 → 100644
浏览文件 @ bf5d80ed
#include "MetaInfo.h"
#include "StrArray.h"
#include <stdio.h>
#include <gtest/gtest.h>
#include <vector>
#include <string>
#define COLN "col1"
#define COLT 253
#define COLL 4000
#define COLS false
#define PK true
#define UK false
#define NN true
#define DEF "default value for col1"
#define ENC "utf-8"
#define SMP(a, b) strcmp((a), (b))
using namespace oceanbase::logmessage;
IColMeta* createColMeta()
{
IColMeta* colMeta = new IColMeta;
colMeta->setName(COLN);
colMeta->setType(COLT);
colMeta->setLength(COLL);
colMeta->setSigned(COLS);
colMeta->setIsPK(PK);
colMeta->setIsUK(UK);
colMeta->setNotNull(NN);
colMeta->setDefault(DEF);
colMeta->setEncoding(ENC);
std::vector<std::string> sets;
sets.push_back("set1");
sets.push_back("set2");
colMeta->setValuesOfEnumSet(sets);
return colMeta;
}
void cmpColMeta(IColMeta& colMeta, std::vector<std::string>& sets)
{
ASSERT_EQ(0, SMP(COLN, colMeta.getName()));
ASSERT_EQ(COLT, colMeta.getType());
ASSERT_EQ(COLL, colMeta.getLength());
ASSERT_EQ(COLS, colMeta.isSigned());
ASSERT_EQ(PK, colMeta.isPK());
ASSERT_EQ(UK, colMeta.isUK());
ASSERT_EQ(NN, colMeta.isNotNull());
ASSERT_EQ(0, SMP(DEF, colMeta.getDefault()));
ASSERT_EQ(0, SMP(ENC, colMeta.getEncoding()));
StrArray* serializedSets = colMeta.getValuesOfEnumSet();
ASSERT_NE((void*)NULL, (void*)serializedSets);
ASSERT_EQ(0, SMP(sets[0].c_str(), (*serializedSets)[0]));
ASSERT_EQ(0, SMP(sets[1].c_str(), (*serializedSets)[1]));
delete serializedSets;
}
TEST(IColMeta, IColMetaAPI)
{
IColMeta* colMeta = createColMeta();
/* Prepare the column meta */
std::vector<std::string> sets;
sets.push_back("set1");
sets.push_back("set2");
cmpColMeta(*colMeta, sets);
/* Compare with parsed one */
std::string s;
ASSERT_EQ(0, colMeta->appendTo(s));
IColMeta parser(s.c_str(), s.size());
size_t msgSize = s.size();
ASSERT_EQ(msgSize, parser.getRealSize());
cmpColMeta(parser, sets);
delete colMeta;
}
#define TN "table1"
ITableMeta* createTableMeta()
{
ITableMeta* tableMeta = new ITableMeta;
tableMeta->setName(TN);
tableMeta->setHasPK(PK);
tableMeta->setHasUK(UK);
tableMeta->setEncoding(ENC);
tableMeta->setDBMeta(NULL);
return tableMeta;
}
void cmpTableMeta(ITableMeta& tableMeta)
{
ASSERT_EQ(0, SMP(TN, tableMeta.getName()));
ASSERT_EQ(PK, tableMeta.hasPK());
ASSERT_EQ(UK, tableMeta.hasUK());
ASSERT_EQ(0, SMP(ENC, tableMeta.getEncoding()));
ASSERT_EQ((void*)NULL, (void*)tableMeta.getDBMeta());
}
TEST(ITableMeta, ITableMetaAPI)
{
// Prepare the table meta
ITableMeta* tableMeta = createTableMeta();
IColMeta* colMeta = createColMeta();
tableMeta->append(colMeta->getName(), colMeta);
// Compare column meta
ASSERT_EQ(colMeta, tableMeta->getCol(0));
ASSERT_EQ(colMeta, tableMeta->getCol(COLN));
std::vector<std::string>& colNames = tableMeta->getColNames();
ASSERT_EQ((size_t)1, colNames.size());
ASSERT_EQ(0, SMP(COLN, colNames[0].c_str()));
// Compare with parsed tablemeta
std::string s;
ASSERT_EQ(0, tableMeta->appendTo(s));
ITableMeta parser(s.c_str(), s.size());
std::string c;
ASSERT_EQ(0, colMeta->appendTo(c));
IColMeta colParser(c.c_str(), c.size());
ASSERT_EQ(s.size(), parser.getRealSize() + colParser.getRealSize());
cmpTableMeta(parser);
delete tableMeta;
}
#define DBN "db1"
IDBMeta* createDBMeta()
{
IDBMeta* dbMeta = new IDBMeta();
dbMeta->setName(DBN);
dbMeta->setEncoding(ENC);
dbMeta->setMetaDataCollections(NULL);
return dbMeta;
}
void cmpDBMeta(IDBMeta* dbMeta)
{
ASSERT_NE((void*)NULL, (void*)dbMeta);
ASSERT_EQ(0, SMP(DBN, dbMeta->getName()));
ASSERT_EQ(0, SMP(ENC, dbMeta->getEncoding()));
ASSERT_EQ((void*)NULL, (void*)dbMeta->getMetaDataCollections());
}
TEST(IDBMeta, IDBMetaAPI)
{
IDBMeta* dbMeta = createDBMeta();
cmpDBMeta(dbMeta);
delete dbMeta;
}
TEST(IMetaDataCollections, IMetaDataCollectionsAPI)
{
IMetaDataCollections* meta = new IMetaDataCollections;
ASSERT_NE((void*)NULL, (void*)meta);
delete meta;
}
int main(int argc, char* argv[])
{
testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}
unittest/memoryleakTest.cpp 0 → 100644
浏览文件 @ bf5d80ed
此差异已折叠。
unittest/varAreaTest.cpp 0 → 100644
浏览文件 @ bf5d80ed
此差异已折叠。
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或