Commit 97849853 authored by Joseph Noir's avatar Joseph Noir

Some cleanup to TCP newbs

parent 5456bca0
...@@ -286,12 +286,8 @@ struct protocol_policy_base { ...@@ -286,12 +286,8 @@ struct protocol_policy_base {
// nop // nop
} }
/// TODO: Come up with something better than a write here?
/// Write header into buffer. Use push back to append only.
virtual void write_header(byte_buffer&, header_writer*) = 0; virtual void write_header(byte_buffer&, header_writer*) = 0;
//virtual size_t offset() const noexcept = 0;
virtual void prepare_for_sending(byte_buffer&, size_t, size_t) = 0; virtual void prepare_for_sending(byte_buffer&, size_t, size_t) = 0;
}; };
...@@ -322,12 +318,6 @@ struct protocol_policy_impl : protocol_policy<typename T::message_type> { ...@@ -322,12 +318,6 @@ struct protocol_policy_impl : protocol_policy<typename T::message_type> {
return impl.read(bytes, count); return impl.read(bytes, count);
} }
/*
size_t offset() const noexcept override {
return T::offset;
}
*/
error timeout(atom_value atm, uint32_t id) override { error timeout(atom_value atm, uint32_t id) override {
return impl.timeout(atm, id); return impl.timeout(atm, id);
} }
...@@ -345,7 +335,9 @@ struct protocol_policy_impl : protocol_policy<typename T::message_type> { ...@@ -345,7 +335,9 @@ struct protocol_policy_impl : protocol_policy<typename T::message_type> {
/// @relates newb /// @relates newb
/// Returned by funtion wr_buf of newb. /// Returned by funtion wr_buf of newb.
template <class Message>
struct write_handle { struct write_handle {
newb<Message>* parent;
protocol_policy_base* protocol; protocol_policy_base* protocol;
byte_buffer* buf; byte_buffer* buf;
size_t header_start; size_t header_start;
...@@ -355,6 +347,7 @@ struct write_handle { ...@@ -355,6 +347,7 @@ struct write_handle {
// Can we calculate added bytes for datagram things? // Can we calculate added bytes for datagram things?
auto payload_size = buf->size() - (header_start + header_len); auto payload_size = buf->size() - (header_start + header_len);
protocol->prepare_for_sending(*buf, header_start, payload_size); protocol->prepare_for_sending(*buf, header_start, payload_size);
parent->flush();
} }
}; };
...@@ -498,7 +491,7 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template ...@@ -498,7 +491,7 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
passivate(); passivate();
} }
write_handle wr_buf(header_writer* hw) { write_handle<Message> wr_buf(header_writer* hw) {
// TODO: We somehow need to tell the transport policy how much we've // TODO: We somehow need to tell the transport policy how much we've
// written to enable it to split the buffer into datagrams. // written to enable it to split the buffer into datagrams.
auto& buf = transport->wr_buf(); auto& buf = transport->wr_buf();
...@@ -507,7 +500,7 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template ...@@ -507,7 +500,7 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
auto hlen = buf.size() - hstart; auto hlen = buf.size() - hstart;
CAF_MESSAGE("returning write buffer starting at " << hstart << " and " CAF_MESSAGE("returning write buffer starting at " << hstart << " and "
<< hlen << " bytes of header"); << hlen << " bytes of header");
return {protocol.get(), &buf, hstart, hlen}; return {this, protocol.get(), &buf, hstart, hlen};
} }
void flush() { void flush() {
...@@ -1073,7 +1066,7 @@ struct tcp_transport_policy : public transport_policy { ...@@ -1073,7 +1066,7 @@ struct tcp_transport_policy : public transport_policy {
} }
byte_buffer& wr_buf() { byte_buffer& wr_buf() {
return send_buffer; return offline_buffer;
} }
void flush(network::event_handler* parent) override { void flush(network::event_handler* parent) override {
...@@ -1109,12 +1102,6 @@ struct tcp_protocol_policy : protocol_policy<typename T::message_type> { ...@@ -1109,12 +1102,6 @@ struct tcp_protocol_policy : protocol_policy<typename T::message_type> {
return impl.read(bytes, count); return impl.read(bytes, count);
} }
/*
size_t offset() const noexcept override {
return T::offset;
}
*/
error timeout(atom_value atm, uint32_t id) override { error timeout(atom_value atm, uint32_t id) override {
return impl.timeout(atm, id); return impl.timeout(atm, id);
} }
...@@ -1159,16 +1146,11 @@ struct tcp_basp_newb : newb<new_tcp_basp_message> { ...@@ -1159,16 +1146,11 @@ struct tcp_basp_newb : newb<new_tcp_basp_message> {
bs(tcp_basp_header{0, sender, receiver}); bs(tcp_basp_header{0, sender, receiver});
return none; return none;
}); });
{ auto whdl = wr_buf(&hw);
// TODO: Need a better idea how to do this ... Maybe pass the write CAF_CHECK(whdl.buf != nullptr);
// handle to flush which then calls `perpare_for_sending`? CAF_CHECK(whdl.protocol != nullptr);
auto whdl = wr_buf(&hw); binary_serializer bs(&backend(), *whdl.buf);
CAF_CHECK(whdl.buf != nullptr); bs(payload);
CAF_CHECK(whdl.protocol != nullptr);
binary_serializer bs(&backend(), *whdl.buf);
bs(payload);
}
flush();
}, },
[=](quit_atom) { [=](quit_atom) {
CAF_MESSAGE("newb actor shutting down"); CAF_MESSAGE("newb actor shutting down");
...@@ -1627,7 +1609,7 @@ CAF_TEST_FIXTURE_SCOPE_END() ...@@ -1627,7 +1609,7 @@ CAF_TEST_FIXTURE_SCOPE_END()
CAF_TEST_FIXTURE_SCOPE(tcp_newbs, fixture) CAF_TEST_FIXTURE_SCOPE(tcp_newbs, fixture)
CAF_TEST(accept test) { CAF_TEST(tcp basp newb) {
scoped_actor main_actor{sys}; scoped_actor main_actor{sys};
actor newb_actor; actor newb_actor;
auto testing = [&](stateful_broker<test_broker_state>* self, auto testing = [&](stateful_broker<test_broker_state>* self,
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment