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
|
#pragma once
#include "remote.hpp"
#include <forstio/buffer.hpp>
#include <forstio/remote/transfer.hpp>
#include <fstream>
#include <cstring>
namespace saw {
template<typename Schema, typename Encoding>
class data_server<Schema, Encoding, rmt::File> final : public i_data_server<rmt::File> {
private:
ptr<remote<rmt::File>> remote_;
remote_address<rmt::File> addr_;
static error generate_invalid_id_error() {
return make_error<err::invalid_state>("rmt::File data_server only supports Id 0.");
}
public:
static constexpr std::pair<uint32_t,uint32_t> class_id{schema_hash<Schema>::apply(), schema_hash<Encoding>::apply()};
data_server(ptr<remote<rmt::File>> remote__, ref<remote_address<rmt::File>> addr__):
remote_{remote__},
addr_{addr__()}
{}
~data_server(){
remote_().deregister_data_server(addr_.get_path_string());
}
SAW_FORBID_MOVE(data_server);
SAW_FORBID_COPY(data_server);
std::pair<uint32_t,uint32_t> get_class_id() const override {
return class_id;
}
error_or<void> send(data<Schema, Encoding>& dat, id<Schema> store_id){
if(store_id.get_value() > 0u){
return generate_invalid_id_error();
}
std::fstream s{addr_.get_path(), s.binary | s.out | s.trunc};
if(!s.is_open()){
return make_error<err::not_found>("Couldn't open file");
}
auto& buff = dat.get_buffer();
buffer_view buff_v{buff};
while(buff_v.read_segment_length() > 0u){
s.write(reinterpret_cast<char*>(&buff_v.read()), buff_v.read_segment_length());
buff_v.read_advance(buff_v.read_segment_length());
}
s.flush();
if(s.bad()){
return make_error<err::invalid_state>("Stream failed to write.");
}
if(s.fail()){
return make_error<err::invalid_state>("Stream encountered non-integer data.");
}
return make_void();
}
error_or<void> allocate(data<typename meta_schema<Schema>::MetaSchema, Encoding> meta, id<Schema> store_id){
if(store_id.get_value() > 0u){
return generate_invalid_id_error();
}
return make_error<err::not_implemented>();
}
error_or<void> erase(id<Schema> store_id){
if(store_id.get_value() > 0u){
return generate_invalid_id_error();
}
return make_error<err::not_implemented>();
}
error_or<data<Schema,Encoding>> receive(id<Schema> store_id){
if(store_id.get_value() > 0u){
return generate_invalid_id_error();
}
data<Schema, Encoding> dat{heap<array_buffer>(1u)};
std::fstream s{addr_.get_path(), s.binary | s.in};
if(!s.is_open()){
return make_error<err::not_found>("Couldn't open file");
}
uint8_t ele{};
while(s.readsome(reinterpret_cast<char*>(&ele), 1u) > 0u){
auto err = dat.get_buffer().push(ele,1u);
if(err.failed()){
return err;
}
}
return dat;
}
};
template <typename Schema, typename Encoding>
class data_client<Schema, Encoding, rmt::File> {
private:
ptr<remote<rmt::File>> remote_;
ptr<data_server<Schema, Encoding, rmt::File>> srv_;
public:
data_client(ptr<remote<rmt::File>> remote__, ptr<data_server<Schema, Encoding,rmt::File>> srv__):
remote_{remote__},
srv_{srv__}
{}
error_or<id<Schema>> send(data<Schema, Encoding>& dat){
auto eov = srv_().send(dat, id<Schema>{0u});
if(eov.is_error()){
return std::move(eov.get_error());
}
return id<Schema>{0u};
}
error_or<data<Schema, Encoding>> receive(id<Schema> id_ = {0u}){
return srv_().receive(id_);
}
};
}
|
