1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
|
/* Copyright (c) 2018, 2025, Oracle and/or its affiliates.
This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License, version 2.0,
as published by the Free Software Foundation.
This program is designed to work with certain software (including
but not limited to OpenSSL) that is licensed under separate terms,
as designated in a particular file or component or in included license
documentation. The authors of MySQL hereby grant you an additional
permission to link the program and your derivative works with the
separately licensed software that they have either included with
the program or referenced in the documentation.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License, version 2.0, for more details.
You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA */
#ifndef BASIC_OSTREAM_INCLUDED
#define BASIC_OSTREAM_INCLUDED
#include <my_byteorder.h>
#include "libbinlogevents/include/compression/compressor.h"
#include "libbinlogevents/include/nodiscard.h"
#include "my_inttypes.h"
#include "my_sys.h"
#include "sql_string.h"
/**
The abstract class for basic output streams which provides write
operation.
*/
class Basic_ostream {
public:
/**
Write some bytes into the output stream.
When all data is written into the stream successfully, then it return
false. Otherwise, true is returned. It will never returns false when
partial data is written into the stream.
@param[in] buffer Data to be written
@param[in] length Length of the data
@retval false Success.
@retval true Error.
*/
virtual bool write(const unsigned char *buffer, my_off_t length) = 0;
virtual ~Basic_ostream() = default;
};
/**
Truncatable_ostream abstract class provides seek() and truncate() interfaces
to all truncatable output streams.
*/
class Truncatable_ostream : public Basic_ostream {
public:
/**
Truncate some data at the end of the output stream.
@param[in] offset Where the output stream will be truncated to.
@retval false Success
@retval true Error
*/
virtual bool truncate(my_off_t offset) = 0;
/**
Put the write position to a given offset. The offset counts from the
beginning of the file.
@param[in] offset Where the write position will be
@retval false Success
@retval true Error
*/
virtual bool seek(my_off_t offset) = 0;
/**
Flush data.
@retval false Success.
@retval true Error.
*/
virtual bool flush() = 0;
/**
Sync.
@retval false Success
@retval true Error
*/
virtual bool sync() = 0;
~Truncatable_ostream() override = default;
};
/**
An output stream based on IO_CACHE class.
*/
class IO_CACHE_ostream : public Truncatable_ostream {
public:
IO_CACHE_ostream();
IO_CACHE_ostream &operator=(const IO_CACHE_ostream &) = delete;
IO_CACHE_ostream(const IO_CACHE_ostream &) = delete;
~IO_CACHE_ostream() override;
/**
Open the output stream. It opens related file and initialize IO_CACHE.
@param[in] log_file_key The PSI_file_key for this stream
@param[in] file_name The file will be opened
@param[in] flags The flags used by IO_CACHE.
@retval false Success
@retval true Error
*/
bool open(
#ifdef HAVE_PSI_INTERFACE
PSI_file_key log_file_key,
#endif
const char *file_name, myf flags);
/**
Closes the stream. It deinitializes IO_CACHE and close the file
it opened.
@retval false Success
@retval true Error
*/
bool close();
bool write(const unsigned char *buffer, my_off_t length) override;
bool seek(my_off_t offset) override;
bool truncate(my_off_t offset) override;
/**
Flush data to IO_CACHE's file if there is any data in IO_CACHE's buffer.
@retval false Success
@retval true Error
*/
bool flush() override;
/**
Syncs the file to disk. It doesn't check and flush any remaining data still
left in IO_CACHE's buffer. So a call to flush() is necessary in order to
persist all data including the data in buffer.
@retval false Success
@retval true Error
*/
bool sync() override;
private:
IO_CACHE m_io_cache;
};
/**
A basic output stream based on StringBuffer class. It has a stack buffer of
size BUFFER_SIZE. It will allocate memory to create a heap buffer if
data exceeds the size of heap buffer.
*/
template <int BUFFER_SIZE>
class StringBuffer_ostream : public Basic_ostream,
public StringBuffer<BUFFER_SIZE> {
public:
StringBuffer_ostream() = default;
StringBuffer_ostream(const StringBuffer_ostream &) = delete;
StringBuffer_ostream &operator=(const StringBuffer_ostream &) = delete;
bool write(const unsigned char *buffer, my_off_t length) override {
return StringBuffer<BUFFER_SIZE>::append(
reinterpret_cast<const char *>(buffer), length);
}
};
class Compressed_ostream : public Basic_ostream {
private:
using Compressor_t = binary_log::transaction::compression::Compressor;
using Compressor_ptr_t = std::shared_ptr<Compressor_t>;
using Status_t = binary_log::transaction::compression::Compress_status;
using Managed_buffer_sequence_t = Compressor_t::Managed_buffer_sequence_t;
Compressor_ptr_t m_compressor;
Managed_buffer_sequence_t &m_managed_buffer_sequence;
Status_t m_status = Status_t::success;
public:
Compressed_ostream(Compressor_ptr_t compressor,
Managed_buffer_sequence_t &managed_buffer_sequence)
: m_compressor(std::move(compressor)),
m_managed_buffer_sequence(managed_buffer_sequence) {}
~Compressed_ostream() override = default;
Compressed_ostream() = delete;
Compressed_ostream(const Compressed_ostream &) = delete;
Compressed_ostream &operator=(const Compressed_ostream &) = delete;
/// Compress the given bytes into the buffer sequence.
///
/// @note This will consume the input, but may not produce all
/// output; it keeps the compression frame open so that the
/// compressor can make use of patterns across different invocations
/// of the function. The caller has to call Compressor::finish to
/// end the frame.
///
/// @param buffer The input buffer
/// @param length The size of the input buffer
/// @retval false Success
/// @retval true Error
[[NODISCARD]] bool write(const unsigned char *buffer,
my_off_t length) override;
Status_t get_status() const;
};
#endif // BASIC_OSTREAM_INCLUDED
|
