9#define LOG(x) do {if (log) *log << x;} while (false)
10#define LOGID(x) do {if (log) *log << get_time_string_of_now() << ' ' << get_session_id() << ": " << x;} while (false)
19 buffer.write<
char>(
'r');
20 buffer.write<int64_t>(offset);
21 buffer.write<int64_t>(offset + int64_t(size));
24 lock.write(buffer.data, buffer.index);
25 lock.read(buffer.data, 9);
28 const int64_t until = buffer.read<int64_t>();
30 if (until < offset || until > offset + int64_t(size))
31 throw Exception(
"bad pread size from server");
33 const size_t returned_size = size_t(until - offset);
35 for (
size_t read = 0; read < returned_size;)
36 read += lock.read_some(data + read, returned_size - read);
50 return server_checkpoint;
52 LOGID(
"checking_hash... ");
54 const int64_t checkpoint = std::min
57 client_journal.get_checkpoint()
62 buffer.write<int64_t>(checkpoint);
72 lock.
write(buffer.data, buffer.index);
73 lock.
read(buffer.data, 1);
76 LOG(buffer.data[0] <<
'\n');
78 if (buffer.data[0] ==
'h')
81 return server_checkpoint;
91 std::chrono::milliseconds wait
96 const char pull_type = char(
'D' +
int(lock_action) + 2*
int(data_transfer));
98 LOGID(
"pulling, lock = " <<
int(lock_action) <<
", data = " <<
99 int(data_transfer) <<
", wait = " <<
double(wait.count()) * 0.001 <<
"s\n");
102 buffer.write<
char>(pull_type);
103 buffer.write<int64_t>(wait.count());
104 buffer.write<int64_t>(client_journal.get_checkpoint());
105 lock.
write(buffer.data, buffer.index);
108 lock.
read(buffer.data, 9);
110 const char reply = buffer.read<
char>();
112 throw Exception(
"pull error: server is pull-only, cannot lock");
113 else if (reply != pull_type)
114 throw Exception(
"pull error: unexpected server reply");
116 server_checkpoint = buffer.read<int64_t>();
118 if (
bool(data_transfer))
121 const int64_t size = server_checkpoint - client_journal.get_checkpoint();
122 Async_Writer writer = client_journal.get_async_tail_writer();
123 LOGID(
"data transfer: ");
124 download(writer, lock, size);
125 client_journal.soft_checkpoint_at(writer.
get_position());
130 return server_checkpoint;
145 const char push_type = char(
'N' +
int(unlock_action));
147 buffer.write<
char>(push_type);
148 buffer.write<int64_t>(from);
149 buffer.write<int64_t>(until);
152 Async_Reader reader = client_journal.get_async_reader(from, until);
154 LOGID(
"pushing(" << push_type <<
")... from = " << from <<
", until = " << until);
157 size_t offset = buffer.index;
161 const size_t size = reader.
read(buffer.data + offset, buffer.size - offset);
163 throw Exception(
"push error: unexpected end of file");
164 lock.
write(buffer.data, size + offset);
170 lock.
read(buffer.data, 1);
172 if (buffer.data[0] ==
'R')
173 throw Exception(
"push error: server is pull-only");
174 else if (buffer.data[0] ==
'C')
176 else if (buffer.data[0] ==
't')
178 else if (buffer.data[0] != push_type)
179 throw Exception(
"push error: unexpected reply");
181 server_checkpoint = until;
183 return server_checkpoint;
190 LOGID(
"releasing lock... ");
202 throw Exception(
"unlock error: unexpected reply");
int64_t get_remaining() const
size_t read(char *buffer, size_t capacity)
bool is_end_of_file() const
int64_t get_position() const
char data[size+extra_size]
void write(const char *data, size_t size)
void read(char *data, size_t size)
static SHA_256::Hash get_full_hash(const Readonly_Journal &journal, int64_t checkpoint)
static SHA_256::Hash get_fast_hash(const Readonly_Journal &journal, int64_t checkpoint)
void print_remaining(int64_t remaining)
Thread_Safe_Channel channel
int64_t pull(Lock_Action lock_action, Data_Transfer data_transfer, Writable_Journal &client_journal, std::chrono::milliseconds wait=std::chrono::milliseconds(0)) override
Pull from the connection.
void unlock() override
Unlock the connection.
int64_t handshake(const Readonly_Journal &client_journal, Content_Check content_check) override
Called during Client construction.
size_t pread(char *data, size_t size, int64_t offset) const
int64_t push(const Readonly_Journal &client_journal, int64_t from, int64_t until, Unlock_Action unlock_action) override
Push new data to the connection.