Skip to content
Projects
Groups
Snippets
Help
Loading...
Help
Support
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
A
actor-incubator
Project overview
Project overview
Details
Activity
Releases
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Issues
0
Issues
0
List
Boards
Labels
Milestones
Merge Requests
0
Merge Requests
0
CI / CD
CI / CD
Pipelines
Jobs
Schedules
Analytics
Analytics
CI / CD
Repository
Value Stream
Wiki
Wiki
Snippets
Snippets
Members
Members
Collapse sidebar
Close sidebar
Activity
Graph
Create a new issue
Jobs
Commits
Issue Boards
Open sidebar
cpp-libs
actor-incubator
Commits
90aa1d31
Commit
90aa1d31
authored
Sep 08, 2019
by
Dominik Charousset
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Implement scaffold for resolving paths
parent
820d1926
Changes
5
Hide whitespace changes
Inline
Side-by-side
Showing
5 changed files
with
181 additions
and
16 deletions
+181
-16
libcaf_net/caf/net/basp/application.hpp
libcaf_net/caf/net/basp/application.hpp
+47
-9
libcaf_net/caf/net/basp/ec.hpp
libcaf_net/caf/net/basp/ec.hpp
+2
-1
libcaf_net/src/application.cpp
libcaf_net/src/application.cpp
+93
-5
libcaf_net/src/ec.cpp
libcaf_net/src/ec.cpp
+1
-0
libcaf_net/test/application.cpp
libcaf_net/test/application.cpp
+38
-1
No files found.
libcaf_net/caf/net/basp/application.hpp
View file @
90aa1d31
...
...
@@ -18,12 +18,14 @@
#pragma once
#include <cstdint>
#include <unordered_map>
#include <unordered_set>
#include <vector>
#include "caf/actor_addr.hpp"
#include "caf/byte.hpp"
#include "caf/callback.hpp"
#include "caf/error.hpp"
#include "caf/net/basp/connection_state.hpp"
#include "caf/net/basp/constants.hpp"
...
...
@@ -33,6 +35,7 @@
#include "caf/node_id.hpp"
#include "caf/serializer_impl.hpp"
#include "caf/span.hpp"
#include "caf/unit.hpp"
namespace
caf
{
namespace
net
{
...
...
@@ -44,6 +47,10 @@ public:
using
buffer_type
=
std
::
vector
<
byte
>
;
using
byte_span
=
span
<
const
byte
>
;
using
write_packet_callback
=
callback
<
byte_span
,
byte_span
>
;
// -- interface functions ----------------------------------------------------
template
<
class
Parent
>
...
...
@@ -55,7 +62,7 @@ public:
return
err
;
auto
hdr
=
to_bytes
(
header
{
message_type
::
handshake
,
static_cast
<
uint32_t
>
(
buf_
.
size
()),
version
});
parent
.
write_packet
(
hdr
,
buf_
);
parent
.
write_packet
(
parent
,
hdr
,
buf_
,
unit
);
return
none
;
}
...
...
@@ -66,17 +73,25 @@ public:
static_cast
<
uint32_t
>
(
ptr
->
payload
.
size
()),
ptr
->
msg
->
mid
.
integer_value
()};
auto
bytes
=
to_bytes
(
hdr
);
parent
.
write_packet
(
make_span
(
bytes
),
ptr
->
payload
);
parent
.
write_packet
(
parent
,
make_span
(
bytes
),
ptr
->
payload
,
unit
);
}
template
<
class
Parent
>
error
handle_data
(
Parent
&
,
span
<
const
byte
>
bytes
)
{
return
handle
(
bytes
);
error
handle_data
(
Parent
&
parent
,
byte_span
bytes
)
{
auto
write_packet
=
make_callback
([
&
](
byte_span
hdr
,
byte_span
payload
)
{
parent
.
write_packet
(
parent
,
hdr
,
payload
,
unit
);
return
none
;
});
return
handle
(
write_packet
,
bytes
);
}
template
<
class
Parent
>
void
resolve
(
Parent
&
,
const
std
::
string
&
,
actor
)
{
// TODO: implement me
void
resolve
(
Parent
&
parent
,
string_view
path
,
actor
listener
)
{
auto
write_packet
=
make_callback
([
&
](
byte_span
hdr
,
byte_span
payload
)
{
parent
.
write_packet
(
parent
,
hdr
,
payload
,
unit
);
return
none
;
});
resolve_remote_path
(
write_packet
,
path
,
listener
);
}
template
<
class
Transport
>
...
...
@@ -91,6 +106,15 @@ public:
static
expected
<
std
::
vector
<
byte
>>
serialize
(
actor_system
&
sys
,
const
type_erased_tuple
&
x
);
// -- utility functions ------------------------------------------------------
strong_actor_ptr
resolve_local_path
(
string_view
path
);
void
resolve_remote_path
(
write_packet_callback
&
write_packet
,
string_view
path
,
actor
listener
);
// -- properties -------------------------------------------------------------
connection_state
state
()
const
noexcept
{
return
state_
;
}
...
...
@@ -102,11 +126,19 @@ public:
private:
// -- message handling -------------------------------------------------------
error
handle
(
span
<
const
byte
>
bytes
);
error
handle
(
write_packet_callback
&
write_packet
,
byte_span
bytes
);
error
handle
(
write_packet_callback
&
write_packet
,
header
hdr
,
byte_span
payload
);
error
handle_handshake
(
write_packet_callback
&
write_packet
,
header
hdr
,
byte_span
payload
);
error
handle
(
header
hdr
,
span
<
const
byte
>
payload
);
error
handle_resolve_request
(
write_packet_callback
&
write_packet
,
header
hdr
,
byte_span
payload
);
error
handle_handshake
(
header
hdr
,
span
<
const
byte
>
payload
);
error
handle_resolve_response
(
write_packet_callback
&
write_packet
,
header
hdr
,
byte_span
payload
);
/// Writes the handshake payload to `buf_`.
error
generate_handshake
();
...
...
@@ -133,6 +165,12 @@ private:
/// Keeps track of which local actors our peer monitors.
std
::
unordered_set
<
actor_addr
>
monitored_actors_
;
/// Caches actor handles obtained via `resolve`.
std
::
unordered_map
<
uint64_t
,
response_promise
>
pending_resolves_
;
/// Ascending ID generator for requests to our peer.
uint64_t
next_request_id_
=
1
;
};
}
// namespace basp
...
...
libcaf_net/caf/net/basp/ec.hpp
View file @
90aa1d31
...
...
@@ -38,8 +38,9 @@ enum class ec : uint8_t {
missing_handshake
,
unexpected_handshake
,
version_mismatch
,
unimplemented
,
unimplemented
=
10
,
app_identifiers_mismatch
,
invalid_payload
,
};
/// @relates ec
...
...
libcaf_net/src/application.cpp
View file @
90aa1d31
...
...
@@ -30,7 +30,9 @@
#include "caf/expected.hpp"
#include "caf/net/basp/constants.hpp"
#include "caf/net/basp/ec.hpp"
#include "caf/no_stages.hpp"
#include "caf/none.hpp"
#include "caf/sec.hpp"
#include "caf/serializer_impl.hpp"
#include "caf/type_erased_tuple.hpp"
...
...
@@ -47,7 +49,32 @@ expected<std::vector<byte>> application::serialize(actor_system& sys,
return
result
;
}
error
application
::
handle
(
span
<
const
byte
>
bytes
)
{
strong_actor_ptr
application
::
resolve_local_path
(
string_view
)
{
return
nullptr
;
}
void
application
::
resolve_remote_path
(
write_packet_callback
&
write_packet
,
string_view
path
,
actor
listener
)
{
buf_
.
clear
();
serializer_impl
<
buffer_type
>
sink
{
system
(),
buf_
};
if
(
auto
err
=
sink
(
path
))
{
CAF_LOG_ERROR
(
"unable to serialize path"
);
return
;
}
auto
req_id
=
next_request_id_
++
;
auto
hdr
=
to_bytes
(
header
{
message_type
::
resolve_request
,
static_cast
<
uint32_t
>
(
buf_
.
size
()),
req_id
});
if
(
auto
err
=
write_packet
(
hdr
,
buf_
))
{
CAF_LOG_ERROR
(
"unable to serialize path"
);
return
;
}
response_promise
rp
{
nullptr
,
actor_cast
<
strong_actor_ptr
>
(
listener
),
no_stages
,
make_message_id
()};
pending_resolves_
.
emplace
(
req_id
,
std
::
move
(
rp
));
}
error
application
::
handle
(
write_packet_callback
&
write_packet
,
byte_span
bytes
)
{
switch
(
state_
)
{
case
connection_state
:
:
await_handshake_header
:
{
if
(
bytes
.
size
()
!=
header_size
)
...
...
@@ -85,7 +112,7 @@ error application::handle(span<const byte> bytes) {
return
ec
::
unexpected_number_of_bytes
;
hdr_
=
header
::
from_bytes
(
bytes
);
if
(
hdr_
.
payload_len
==
0
)
return
handle
(
hdr_
,
span
<
const
byte
>
{});
return
handle
(
write_packet
,
hdr_
,
byte_span
{});
else
state_
=
connection_state
::
await_payload
;
return
none
;
...
...
@@ -93,17 +120,23 @@ error application::handle(span<const byte> bytes) {
case
connection_state
:
:
await_payload
:
{
if
(
bytes
.
size
()
!=
hdr_
.
payload_len
)
return
ec
::
unexpected_number_of_bytes
;
return
handle
(
hdr_
,
bytes
);
state_
=
connection_state
::
await_header
;
return
handle
(
write_packet
,
hdr_
,
bytes
);
}
default:
return
ec
::
illegal_state
;
}
}
error
application
::
handle
(
header
hdr
,
span
<
const
byte
>
)
{
error
application
::
handle
(
write_packet_callback
&
write_packet
,
header
hdr
,
byte_span
payload
)
{
switch
(
hdr
.
type
)
{
case
message_type
:
:
handshake
:
return
ec
::
unexpected_handshake
;
case
message_type
:
:
resolve_request
:
return
handle_resolve_request
(
write_packet
,
hdr
,
payload
);
case
message_type
:
:
resolve_response
:
return
handle_resolve_response
(
write_packet
,
hdr
,
payload
);
case
message_type
:
:
heartbeat
:
return
none
;
default:
...
...
@@ -111,7 +144,8 @@ error application::handle(header hdr, span<const byte>) {
}
}
error
application
::
handle_handshake
(
header
hdr
,
span
<
const
byte
>
payload
)
{
error
application
::
handle_handshake
(
write_packet_callback
&
,
header
hdr
,
byte_span
payload
)
{
if
(
hdr
.
type
!=
message_type
::
handshake
)
return
ec
::
missing_handshake
;
if
(
hdr
.
operation_data
!=
version
)
...
...
@@ -127,6 +161,60 @@ error application::handle_handshake(header hdr, span<const byte> payload) {
return
none
;
}
error
application
::
handle_resolve_request
(
write_packet_callback
&
write_packet
,
header
hdr
,
byte_span
payload
)
{
CAF_ASSERT
(
hdr
.
type
==
message_type
::
resolve_request
);
size_t
path_size
=
0
;
binary_deserializer
source
{
system
(),
payload
};
if
(
auto
err
=
source
.
begin_sequence
(
path_size
))
return
err
;
// We expect the payload to consist only of the path.
if
(
path_size
!=
source
.
remaining
())
return
ec
::
invalid_payload
;
auto
remainder
=
source
.
remainder
();
string_view
path
{
reinterpret_cast
<
const
char
*>
(
remainder
.
data
()),
remainder
.
size
()};
// Write result.
auto
result
=
resolve_local_path
(
path
);
buf_
.
clear
();
actor_id
aid
=
result
?
result
->
id
()
:
0
;
std
::
set
<
std
::
string
>
ifs
;
// TODO: figure out how to obtain messaging interface.
serializer_impl
<
buffer_type
>
sink
{
system
(),
buf_
};
if
(
auto
err
=
sink
(
aid
,
ifs
))
return
err
;
auto
out_hdr
=
to_bytes
(
header
{
message_type
::
resolve_response
,
static_cast
<
uint32_t
>
(
buf_
.
size
()),
hdr
.
operation_data
});
return
write_packet
(
out_hdr
,
buf_
);
}
error
application
::
handle_resolve_response
(
write_packet_callback
&
,
header
hdr
,
byte_span
payload
)
{
CAF_ASSERT
(
hdr
.
type
==
message_type
::
resolve_response
);
auto
i
=
pending_resolves_
.
find
(
hdr
.
operation_data
);
if
(
i
==
pending_resolves_
.
end
())
{
CAF_LOG_ERROR
(
"received unknown ID in resolve_response message"
);
return
none
;
}
auto
guard
=
detail
::
make_scope_guard
([
&
]
{
if
(
i
->
second
.
pending
())
i
->
second
.
deliver
(
sec
::
remote_lookup_failed
);
pending_resolves_
.
erase
(
i
);
});
actor_id
aid
;
std
::
set
<
std
::
string
>
ifs
;
binary_deserializer
source
{
system
(),
payload
};
if
(
auto
err
=
source
(
aid
,
ifs
))
return
err
;
if
(
aid
==
0
)
{
i
->
second
.
deliver
(
strong_actor_ptr
{
nullptr
});
return
none
;
}
// TODO: generate proxies and deal with proxy_registry
return
ec
::
unimplemented
;
}
error
application
::
generate_handshake
()
{
serializer_impl
<
buffer_type
>
sink
{
system
(),
buf_
};
return
sink
(
system
().
node
(),
...
...
libcaf_net/src/ec.cpp
View file @
90aa1d31
...
...
@@ -39,6 +39,7 @@ const char* ec_names[] = {
"version_mismatch"
,
"unimplemented"
,
"app_identifiers_mismatch"
,
"invalid_payload"
,
};
}
// namespace
...
...
libcaf_net/test/application.cpp
View file @
90aa1d31
...
...
@@ -63,7 +63,8 @@ struct fixture : test_coordinator_fixture<> {
input
=
to_buf
(
xs
...);
}
void
write_packet
(
span
<
const
byte
>
hdr
,
span
<
const
byte
>
payload
)
{
void
write_packet
(
fixture
&
,
span
<
const
byte
>
hdr
,
span
<
const
byte
>
payload
,
unit_t
)
{
output
.
insert
(
output
.
end
(),
hdr
.
begin
(),
hdr
.
end
());
output
.
insert
(
output
.
end
(),
payload
.
begin
(),
payload
.
end
());
}
...
...
@@ -114,6 +115,29 @@ struct fixture : test_coordinator_fixture<> {
}
// namespace
#define MOCK(kind, op, ...) \
do { \
auto payload = to_buf(__VA_ARGS__); \
set_input(basp::header{kind, static_cast<uint32_t>(payload.size()), op}); \
if (auto err = app.handle_data(*this, input)) \
CAF_FAIL("application-under-test failed to process header: " \
<< sys.render(err)); \
if (auto err = app.handle_data(*this, payload)) \
CAF_FAIL("application-under-test failed to process payload: " \
<< sys.render(err)); \
} while (false)
#define RECEIVE(msg_type, op_data, ...) \
do { \
serializer_impl<buffer_type> source{sys, output}; \
basp::header hdr; \
if (auto err = source(hdr, __VA_ARGS__)) \
CAF_FAIL("failed to receive data: " << sys.render(err)); \
CAF_CHECK_EQUAL(hdr.type, msg_type); \
CAF_CHECK_EQUAL(hdr.operation_data, op_data); \
output.clear(); \
} while (false)
CAF_TEST_FIXTURE_SCOPE
(
application_tests
,
fixture
)
CAF_TEST
(
missing
handshake
)
{
...
...
@@ -172,6 +196,19 @@ CAF_TEST(repeated handshake) {
basp
::
ec
::
unexpected_handshake
);
}
CAF_TEST
(
resolve
request
without
result
)
{
handle_handshake
();
consume_handshake
();
CAF_CHECK_EQUAL
(
app
.
state
(),
basp
::
connection_state
::
await_header
);
MOCK
(
basp
::
message_type
::
resolve_request
,
42
,
std
::
string
{
"/foo/bar"
});
CAF_CHECK_EQUAL
(
app
.
state
(),
basp
::
connection_state
::
await_header
);
actor_id
aid
;
std
::
set
<
std
::
string
>
ifs
;
RECEIVE
(
basp
::
message_type
::
resolve_response
,
42u
,
aid
,
ifs
);
CAF_CHECK_EQUAL
(
aid
,
0u
);
CAF_CHECK
(
ifs
.
empty
());
}
CAF_TEST
(
heartbeat
message
)
{
handle_handshake
();
consume_handshake
();
...
...
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment