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
56fb469a
Commit
56fb469a
authored
Jul 30, 2019
by
Jakob Otto
Browse files
Options
Browse Files
Download
Plain Diff
Merge branch 'master' into topic/scribe
parents
245ca656
5e32089a
Changes
11
Hide whitespace changes
Inline
Side-by-side
Showing
11 changed files
with
417 additions
and
31 deletions
+417
-31
libcaf_net/CMakeLists.txt
libcaf_net/CMakeLists.txt
+2
-0
libcaf_net/caf/net/actor_proxy_impl.hpp
libcaf_net/caf/net/actor_proxy_impl.hpp
+50
-0
libcaf_net/caf/net/datagram_socket.hpp
libcaf_net/caf/net/datagram_socket.hpp
+53
-0
libcaf_net/caf/net/endpoint_manager.hpp
libcaf_net/caf/net/endpoint_manager.hpp
+19
-6
libcaf_net/caf/net/endpoint_manager_impl.hpp
libcaf_net/caf/net/endpoint_manager_impl.hpp
+22
-16
libcaf_net/caf/net/socket.hpp
libcaf_net/caf/net/socket.hpp
+1
-1
libcaf_net/src/actor_proxy_impl.cpp
libcaf_net/src/actor_proxy_impl.cpp
+72
-0
libcaf_net/src/datagram_socket.cpp
libcaf_net/src/datagram_socket.cpp
+64
-0
libcaf_net/src/endpoint_manager.cpp
libcaf_net/src/endpoint_manager.cpp
+30
-1
libcaf_net/test/datagram_socket.cpp
libcaf_net/test/datagram_socket.cpp
+31
-0
libcaf_net/test/endpoint_manager.cpp
libcaf_net/test/endpoint_manager.cpp
+73
-7
No files found.
libcaf_net/CMakeLists.txt
View file @
56fb469a
...
@@ -7,6 +7,8 @@ file(GLOB_RECURSE LIBCAF_NET_HDRS "caf/*.hpp")
...
@@ -7,6 +7,8 @@ file(GLOB_RECURSE LIBCAF_NET_HDRS "caf/*.hpp")
# list cpp files excluding platform-dependent files
# list cpp files excluding platform-dependent files
set
(
LIBCAF_NET_SRCS
set
(
LIBCAF_NET_SRCS
src/actor_proxy_impl.cpp
src/datagram_socket.cpp
src/endpoint_manager.cpp
src/endpoint_manager.cpp
src/host.cpp
src/host.cpp
src/multiplexer.cpp
src/multiplexer.cpp
...
...
libcaf_net/caf/net/actor_proxy_impl.hpp
0 → 100644
View file @
56fb469a
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include "caf/actor_proxy.hpp"
#include "caf/net/endpoint_manager.hpp"
namespace
caf
{
namespace
net
{
/// Implements a simple proxy forwarding all operations to a manager.
class
actor_proxy_impl
:
public
actor_proxy
{
public:
using
super
=
actor_proxy
;
actor_proxy_impl
(
actor_config
&
cfg
,
endpoint_manager_ptr
dst
);
~
actor_proxy_impl
()
override
;
void
enqueue
(
mailbox_element_ptr
what
,
execution_unit
*
context
)
override
;
bool
add_backlink
(
abstract_actor
*
x
)
override
;
bool
remove_backlink
(
abstract_actor
*
x
)
override
;
void
kill_proxy
(
execution_unit
*
ctx
,
error
rsn
)
override
;
private:
endpoint_manager
::
serialize_fun_type
sf_
;
endpoint_manager_ptr
dst_
;
};
}
// namespace net
}
// namespace caf
libcaf_net/caf/net/datagram_socket.hpp
0 → 100644
View file @
56fb469a
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include "caf/net/network_socket.hpp"
#include "caf/variant.hpp"
namespace
caf
{
namespace
net
{
/// A datagram-oriented network communication endpoint.
struct
datagram_socket
:
abstract_socket
<
datagram_socket
>
{
using
super
=
abstract_socket
<
datagram_socket
>
;
using
super
::
super
;
constexpr
operator
socket
()
const
noexcept
{
return
socket
{
id
};
}
constexpr
operator
network_socket
()
const
noexcept
{
return
network_socket
{
id
};
}
};
/// Enables or disables `SIO_UDP_CONNRESET` error on `x`.
/// @relates datagram_socket
error
allow_connreset
(
datagram_socket
x
,
bool
new_value
);
/// Converts the result from I/O operation on a ::datagram_socket to either an
/// error code or a integer greater or equal to zero.
/// @relates datagram_socket
variant
<
size_t
,
sec
>
check_datagram_socket_io_res
(
std
::
make_signed
<
size_t
>::
type
res
);
}
// namespace net
}
// namespace caf
libcaf_net/caf/net/endpoint_manager.hpp
View file @
56fb469a
...
@@ -41,6 +41,13 @@ public:
...
@@ -41,6 +41,13 @@ public:
using
super
=
socket_manager
;
using
super
=
socket_manager
;
/// Represents either an error or a serialized payload.
using
maybe_buffer
=
expected
<
std
::
vector
<
char
>>
;
/// A function type for serializing message payloads.
using
serialize_fun_type
=
maybe_buffer
(
*
)(
actor_system
&
,
const
type_erased_tuple
&
);
struct
event
:
intrusive
::
singly_linked
<
event
>
{
struct
event
:
intrusive
::
singly_linked
<
event
>
{
struct
resolve_request
{
struct
resolve_request
{
std
::
string
path
;
std
::
string
path
;
...
@@ -84,6 +91,8 @@ public:
...
@@ -84,6 +91,8 @@ public:
/// Serialized representation of of `msg->content()`.
/// Serialized representation of of `msg->content()`.
std
::
vector
<
char
>
payload
;
std
::
vector
<
char
>
payload
;
message
(
mailbox_element_ptr
msg
,
std
::
vector
<
char
>
payload
);
};
};
struct
message_policy
{
struct
message_policy
{
...
@@ -113,24 +122,28 @@ public:
...
@@ -113,24 +122,28 @@ public:
// -- properties -------------------------------------------------------------
// -- properties -------------------------------------------------------------
event_queue_type
&
event_queue
()
{
actor_system
&
system
()
{
return
event
s_
;
return
sy
s_
;
}
}
message_queue_type
&
message_queue
()
{
std
::
unique_ptr
<
message
>
next_message
();
return
messages_
;
}
// -- event management -------------------------------------------------------
// -- event management -------------------------------------------------------
/// Resolves a path to a remote actor
/// Resolves a path to a remote actor
.
void
resolve
(
std
::
string
path
,
actor
listener
);
void
resolve
(
std
::
string
path
,
actor
listener
);
/// Enqueues a message to the endpoint.
void
enqueue
(
mailbox_element_ptr
msg
,
std
::
vector
<
char
>
payload
);
// -- pure virtual member functions ------------------------------------------
// -- pure virtual member functions ------------------------------------------
/// Initializes the manager before adding it to the multiplexer's event loop.
/// Initializes the manager before adding it to the multiplexer's event loop.
virtual
error
init
()
=
0
;
virtual
error
init
()
=
0
;
/// @returns the protocol-specific function for serializing payloads.
virtual
serialize_fun_type
serialize_fun
()
const
noexcept
=
0
;
protected:
protected:
/// Points to the hosting actor system.
/// Points to the hosting actor system.
actor_system
&
sys_
;
actor_system
&
sys_
;
...
...
libcaf_net/caf/net/endpoint_manager_impl.hpp
View file @
56fb469a
...
@@ -69,21 +69,23 @@ public:
...
@@ -69,21 +69,23 @@ public:
}
}
bool
handle_write_event
()
override
{
bool
handle_write_event
()
override
{
if
(
!
this
->
events_
.
empty
())
{
if
(
!
this
->
events_
.
blocked
()
&&
!
this
->
events_
.
empty
())
{
this
->
events_
.
fetch_more
();
do
{
auto
&
q
=
this
->
events_
.
queue
();
this
->
events_
.
fetch_more
();
q
.
inc_deficit
(
q
.
total_task_size
());
auto
&
q
=
this
->
events_
.
queue
();
for
(
auto
ptr
=
q
.
next
();
ptr
!=
nullptr
;
ptr
=
q
.
next
())
{
q
.
inc_deficit
(
q
.
total_task_size
());
using
timeout
=
endpoint_manager
::
event
::
timeout
;
for
(
auto
ptr
=
q
.
next
();
ptr
!=
nullptr
;
ptr
=
q
.
next
())
{
using
resolve_request
=
endpoint_manager
::
event
::
resolve_request
;
using
timeout
=
endpoint_manager
::
event
::
timeout
;
if
(
auto
rr
=
get_if
<
resolve_request
>
(
&
ptr
->
value
))
{
using
resolve_request
=
endpoint_manager
::
event
::
resolve_request
;
transport_
.
resolve
(
*
this
,
std
::
move
(
rr
->
path
),
if
(
auto
rr
=
get_if
<
resolve_request
>
(
&
ptr
->
value
))
{
std
::
move
(
rr
->
listener
));
transport_
.
resolve
(
*
this
,
std
::
move
(
rr
->
path
),
}
else
{
std
::
move
(
rr
->
listener
));
auto
&
t
=
get
<
timeout
>
(
ptr
->
value
);
}
else
{
transport_
.
timeout
(
*
this
,
t
.
type
,
t
.
id
);
auto
&
t
=
get
<
timeout
>
(
ptr
->
value
);
transport_
.
timeout
(
*
this
,
t
.
type
,
t
.
id
);
}
}
}
}
}
while
(
!
this
->
events_
.
try_block
());
}
}
return
transport_
.
handle_write_event
(
*
this
);
return
transport_
.
handle_write_event
(
*
this
);
}
}
...
@@ -92,9 +94,13 @@ public:
...
@@ -92,9 +94,13 @@ public:
transport_
.
handle_error
(
application_
,
code
);
transport_
.
handle_error
(
application_
,
code
);
}
}
serialize_fun_type
serialize_fun
()
const
noexcept
override
{
return
application_type
::
serialize
;
}
private:
private:
Transport
transport_
;
transport_type
transport_
;
Application
application_
;
application_type
application_
;
};
};
}
// namespace net
}
// namespace net
...
...
libcaf_net/caf/net/socket.hpp
View file @
56fb469a
...
@@ -31,7 +31,7 @@ namespace caf {
...
@@ -31,7 +31,7 @@ namespace caf {
namespace
net
{
namespace
net
{
/// An internal endpoint for sending or receiving data. Can be either a
/// An internal endpoint for sending or receiving data. Can be either a
/// ::network_socket
or a ::pipe
_socket.
/// ::network_socket
, ::pipe_socket, ::stream_socket, or ::datagram
_socket.
struct
socket
:
abstract_socket
<
socket
>
{
struct
socket
:
abstract_socket
<
socket
>
{
using
super
=
abstract_socket
<
socket
>
;
using
super
=
abstract_socket
<
socket
>
;
...
...
libcaf_net/src/actor_proxy_impl.cpp
0 → 100644
View file @
56fb469a
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/net/actor_proxy_impl.hpp"
#include "caf/actor_system.hpp"
#include "caf/expected.hpp"
#include "caf/logger.hpp"
namespace
caf
{
namespace
net
{
actor_proxy_impl
::
actor_proxy_impl
(
actor_config
&
cfg
,
endpoint_manager_ptr
dst
)
:
super
(
cfg
),
sf_
(
dst
->
serialize_fun
()),
dst_
(
std
::
move
(
dst
))
{
// anon_send(broker_, monitor_atom::value, ctrl());
}
actor_proxy_impl
::~
actor_proxy_impl
()
{
// anon_send(broker_, make_message(delete_atom::value, node(), id()));
}
void
actor_proxy_impl
::
enqueue
(
mailbox_element_ptr
what
,
execution_unit
*
)
{
CAF_PUSH_AID
(
0
);
CAF_ASSERT
(
what
!=
nullptr
);
if
(
auto
payload
=
sf_
(
home_system
(),
what
->
content
()))
dst_
->
enqueue
(
std
::
move
(
what
),
std
::
move
(
*
payload
));
else
CAF_LOG_ERROR
(
"unable to serialize payload: "
<<
home_system
().
render
(
payload
.
error
()));
}
bool
actor_proxy_impl
::
add_backlink
(
abstract_actor
*
x
)
{
if
(
monitorable_actor
::
add_backlink
(
x
))
{
enqueue
(
make_mailbox_element
(
ctrl
(),
make_message_id
(),
{},
link_atom
::
value
,
x
->
ctrl
()),
nullptr
);
return
true
;
}
return
false
;
}
bool
actor_proxy_impl
::
remove_backlink
(
abstract_actor
*
x
)
{
if
(
monitorable_actor
::
remove_backlink
(
x
))
{
enqueue
(
make_mailbox_element
(
ctrl
(),
make_message_id
(),
{},
unlink_atom
::
value
,
x
->
ctrl
()),
nullptr
);
return
true
;
}
return
false
;
}
void
actor_proxy_impl
::
kill_proxy
(
execution_unit
*
ctx
,
error
rsn
)
{
cleanup
(
std
::
move
(
rsn
),
ctx
);
}
}
// namespace net
}
// namespace caf
libcaf_net/src/datagram_socket.cpp
0 → 100644
View file @
56fb469a
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/net/datagram_socket.hpp"
#include "caf/detail/net_syscall.hpp"
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/logger.hpp"
namespace
caf
{
namespace
net
{
#ifdef CAF_WINDOWS
error
allow_connreset
(
datagram_socket
x
,
bool
new_value
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
x
)
<<
CAF_ARG
(
new_value
));
DWORD
bytes_returned
=
0
;
CAF_NET_SYSCALL
(
"WSAIoctl"
,
res
,
!=
,
0
,
WSAIoctl
(
x
.
id
,
_WSAIOW
(
IOC_VENDOR
,
12
),
&
new_value
,
sizeof
(
new_value
),
NULL
,
0
,
&
bytes_returned
,
NULL
,
NULL
));
return
none
;
}
#else // CAF_WINDOWS
error
allow_connreset
(
datagram_socket
x
,
bool
)
{
if
(
x
==
invalid_socket
)
return
sec
::
socket_invalid
;
// nop; SIO_UDP_CONNRESET only exists on Windows
return
none
;
}
#endif // CAF_WINDOWS
variant
<
size_t
,
sec
>
check_datagram_socket_io_res
(
std
::
make_signed
<
size_t
>::
type
res
)
{
if
(
res
<
0
)
{
auto
code
=
last_socket_error
();
if
(
code
==
std
::
errc
::
operation_would_block
||
code
==
std
::
errc
::
resource_unavailable_try_again
)
return
sec
::
unavailable_or_would_block
;
return
sec
::
socket_operation_failed
;
}
return
static_cast
<
size_t
>
(
res
);
}
}
// namespace net
}
// namespace caf
libcaf_net/src/endpoint_manager.cpp
View file @
56fb469a
...
@@ -35,19 +35,41 @@ endpoint_manager::event::event(atom_value type, uint64_t id)
...
@@ -35,19 +35,41 @@ endpoint_manager::event::event(atom_value type, uint64_t id)
// nop
// nop
}
}
endpoint_manager
::
message
::
message
(
mailbox_element_ptr
msg
,
std
::
vector
<
char
>
payload
)
:
msg
(
std
::
move
(
msg
)),
payload
(
std
::
move
(
payload
))
{
// nop
}
endpoint_manager
::
endpoint_manager
(
socket
handle
,
const
multiplexer_ptr
&
parent
,
endpoint_manager
::
endpoint_manager
(
socket
handle
,
const
multiplexer_ptr
&
parent
,
actor_system
&
sys
)
actor_system
&
sys
)
:
super
(
handle
,
parent
),
:
super
(
handle
,
parent
),
sys_
(
sys
),
sys_
(
sys
),
events_
(
event_policy
{}),
events_
(
event_policy
{}),
messages_
(
message_policy
{})
{
messages_
(
message_policy
{})
{
// nop
events_
.
try_block
();
messages_
.
try_block
();
}
}
endpoint_manager
::~
endpoint_manager
()
{
endpoint_manager
::~
endpoint_manager
()
{
// nop
// nop
}
}
std
::
unique_ptr
<
endpoint_manager
::
message
>
endpoint_manager
::
next_message
()
{
if
(
messages_
.
blocked
())
return
nullptr
;
messages_
.
fetch_more
();
auto
&
q
=
messages_
.
queue
();
auto
ts
=
q
.
next_task_size
();
if
(
ts
==
0
)
return
nullptr
;
q
.
inc_deficit
(
ts
);
auto
result
=
q
.
next
();
if
(
q
.
empty
())
messages_
.
try_block
();
return
result
;
}
void
endpoint_manager
::
resolve
(
std
::
string
path
,
actor
listener
)
{
void
endpoint_manager
::
resolve
(
std
::
string
path
,
actor
listener
)
{
using
intrusive
::
inbox_result
;
using
intrusive
::
inbox_result
;
auto
ptr
=
new
event
(
std
::
move
(
path
),
std
::
move
(
listener
));
auto
ptr
=
new
event
(
std
::
move
(
path
),
std
::
move
(
listener
));
...
@@ -63,5 +85,12 @@ void endpoint_manager::resolve(std::string path, actor listener) {
...
@@ -63,5 +85,12 @@ void endpoint_manager::resolve(std::string path, actor listener) {
}
}
}
}
void
endpoint_manager
::
enqueue
(
mailbox_element_ptr
msg
,
std
::
vector
<
char
>
payload
)
{
auto
ptr
=
new
message
(
std
::
move
(
msg
),
std
::
move
(
payload
));
if
(
messages_
.
push_back
(
ptr
)
==
intrusive
::
inbox_result
::
unblocked_reader
)
mask_add
(
operation
::
write
);
}
}
// namespace net
}
// namespace net
}
// namespace caf
}
// namespace caf
libcaf_net/test/datagram_socket.cpp
0 → 100644
View file @
56fb469a
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#define CAF_SUITE datagram_socket
#include "caf/net/datagram_socket.hpp"
#include "caf/test/dsl.hpp"
using
namespace
caf
;
using
namespace
caf
::
net
;
CAF_TEST
(
invalid_socket
)
{
datagram_socket
x
;
CAF_CHECK_NOT_EQUAL
(
allow_connreset
(
x
,
true
),
none
);
}
libcaf_net/test/endpoint_manager.cpp
View file @
56fb469a
...
@@ -24,7 +24,10 @@
...
@@ -24,7 +24,10 @@
#include "host_fixture.hpp"
#include "host_fixture.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/make_actor.hpp"
#include "caf/net/actor_proxy_impl.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/net/stream_socket.hpp"
...
@@ -45,10 +48,25 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
...
@@ -45,10 +48,25 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
CAF_FAIL
(
"mpx->init failed: "
<<
sys
.
render
(
err
));
CAF_FAIL
(
"mpx->init failed: "
<<
sys
.
render
(
err
));
}
}
bool
handle_io_event
()
override
{
mpx
->
handle_updates
();
return
mpx
->
poll_once
(
false
);
}
multiplexer_ptr
mpx
;
multiplexer_ptr
mpx
;
};
};
class
dummy_application
{};
class
dummy_application
{
public:
static
expected
<
std
::
vector
<
char
>>
serialize
(
actor_system
&
sys
,
const
type_erased_tuple
&
x
)
{
std
::
vector
<
char
>
result
;
binary_serializer
sink
{
sys
,
result
};
if
(
auto
err
=
message
::
save
(
sink
,
x
))
return
err
;
return
result
;
}
};
class
dummy_transport
{
class
dummy_transport
{
public:
public:
...
@@ -80,7 +98,11 @@ public:
...
@@ -80,7 +98,11 @@ public:
}
}
template
<
class
Manager
>
template
<
class
Manager
>
bool
handle_write_event
(
Manager
&
)
{
bool
handle_write_event
(
Manager
&
mgr
)
{
for
(
auto
x
=
mgr
.
next_message
();
x
!=
nullptr
;
x
=
mgr
.
next_message
())
{
auto
&
payload
=
x
->
payload
;
write_buf_
.
insert
(
write_buf_
.
end
(),
payload
.
begin
(),
payload
.
end
());
}
auto
res
=
write
(
handle_
,
write_buf_
.
data
(),
write_buf_
.
size
());
auto
res
=
write
(
handle_
,
write_buf_
.
data
(),
write_buf_
.
size
());
if
(
auto
num_bytes
=
get_if
<
size_t
>
(
&
res
))
{
if
(
auto
num_bytes
=
get_if
<
size_t
>
(
&
res
))
{
write_buf_
.
erase
(
write_buf_
.
begin
(),
write_buf_
.
begin
()
+
*
num_bytes
);
write_buf_
.
erase
(
write_buf_
.
begin
(),
write_buf_
.
begin
()
+
*
num_bytes
);
...
@@ -95,9 +117,14 @@ public:
...
@@ -95,9 +117,14 @@ public:
}
}
template
<
class
Manager
>
template
<
class
Manager
>
void
resolve
(
Manager
&
,
std
::
string
path
,
actor
listener
)
{
void
resolve
(
Manager
&
mgr
,
std
::
string
path
,
actor
listener
)
{
anon_send
(
listener
,
resolve_atom
::
value
,
std
::
move
(
path
),
actor_id
aid
=
42
;
make_error
(
sec
::
feature_disabled
));
node_id
nid
{
42
,
"00112233445566778899aa00112233445566778899aa"
};
actor_config
cfg
;
auto
p
=
make_actor
<
actor_proxy_impl
,
strong_actor_ptr
>
(
aid
,
nid
,
&
mgr
.
system
(),
cfg
,
&
mgr
);
anon_send
(
listener
,
resolve_atom
::
value
,
std
::
move
(
path
),
p
);
}
}
template
<
class
Manager
>
template
<
class
Manager
>
...
@@ -137,12 +164,51 @@ CAF_TEST(send and receive) {
...
@@ -137,12 +164,51 @@ CAF_TEST(send and receive) {
CAF_CHECK_EQUAL
(
write
(
sockets
.
second
,
hello_manager
.
data
(),
CAF_CHECK_EQUAL
(
write
(
sockets
.
second
,
hello_manager
.
data
(),
hello_manager
.
size
()),
hello_manager
.
size
()),
hello_manager
.
size
());
hello_manager
.
size
());
while
(
mpx
->
poll_once
(
false
))
run
();
;
// Repeat.
CAF_CHECK_EQUAL
(
string_view
(
buf
->
data
(),
buf
->
size
()),
hello_manager
);
CAF_CHECK_EQUAL
(
string_view
(
buf
->
data
(),
buf
->
size
()),
hello_manager
);
CAF_CHECK_EQUAL
(
read
(
sockets
.
second
,
read_buf
.
data
(),
read_buf
.
size
()),
CAF_CHECK_EQUAL
(
read
(
sockets
.
second
,
read_buf
.
data
(),
read_buf
.
size
()),
hello_test
.
size
());
hello_test
.
size
());
CAF_CHECK_EQUAL
(
string_view
(
read_buf
.
data
(),
hello_test
.
size
()),
hello_test
);
CAF_CHECK_EQUAL
(
string_view
(
read_buf
.
data
(),
hello_test
.
size
()),
hello_test
);
}
}
CAF_TEST
(
resolve
and
proxy
communication
)
{
std
::
vector
<
char
>
read_buf
(
1024
);
auto
buf
=
std
::
make_shared
<
std
::
vector
<
char
>>
();
auto
sockets
=
unbox
(
make_stream_socket_pair
());
nonblocking
(
sockets
.
second
,
true
);
auto
guard
=
detail
::
make_scope_guard
([
&
]
{
close
(
sockets
.
second
);
});
auto
mgr
=
make_endpoint_manager
(
mpx
,
sys
,
dummy_transport
{
sockets
.
first
,
buf
},
dummy_application
{});
CAF_CHECK_EQUAL
(
mgr
->
init
(),
none
);
mpx
->
handle_updates
();
run
();
CAF_CHECK_EQUAL
(
read
(
sockets
.
second
,
read_buf
.
data
(),
read_buf
.
size
()),
hello_test
.
size
());
mgr
->
resolve
(
"/id/42"
,
self
);
run
();
self
->
receive
(
[
&
](
resolve_atom
,
const
std
::
string
&
,
const
strong_actor_ptr
&
p
)
{
CAF_MESSAGE
(
"got a proxy, send a message to it"
);
self
->
send
(
actor_cast
<
actor
>
(
p
),
"hello proxy!"
);
},
after
(
std
::
chrono
::
seconds
(
0
))
>>
[
&
]
{
CAF_FAIL
(
"manager did not respond with a proxy."
);
});
run
();
auto
read_res
=
read
(
sockets
.
second
,
read_buf
.
data
(),
read_buf
.
size
());
if
(
!
holds_alternative
<
size_t
>
(
read_res
))
{
CAF_ERROR
(
"read() returned an error: "
<<
sys
.
render
(
get
<
sec
>
(
read_res
)));
return
;
}
read_buf
.
resize
(
get
<
size_t
>
(
read_res
));
CAF_MESSAGE
(
"receive buffer contains "
<<
read_buf
.
size
()
<<
" bytes"
);
message
msg
;
binary_deserializer
source
{
sys
,
read_buf
};
CAF_CHECK_EQUAL
(
source
(
msg
),
none
);
if
(
msg
.
match_elements
<
std
::
string
>
())
CAF_CHECK_EQUAL
(
msg
.
get_as
<
std
::
string
>
(
0
),
"hello proxy!"
);
else
CAF_ERROR
(
"expected a string, got: "
<<
to_string
(
msg
));
}
CAF_TEST_FIXTURE_SCOPE_END
()
CAF_TEST_FIXTURE_SCOPE_END
()
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