Skip to content
Projects
Groups
Snippets
Help
Loading...
Help
Support
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
A
Actor Framework
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
Operations
Operations
Metrics
Environments
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 Framework
Commits
412be4ab
Commit
412be4ab
authored
Nov 22, 2019
by
Jakob Otto
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Fix test
parent
930d232a
Changes
6
Hide whitespace changes
Inline
Side-by-side
Showing
6 changed files
with
95 additions
and
46 deletions
+95
-46
libcaf_net/CMakeLists.txt
libcaf_net/CMakeLists.txt
+2
-0
libcaf_net/caf/net/doorman.hpp
libcaf_net/caf/net/doorman.hpp
+48
-13
libcaf_net/caf/net/socket_manager.hpp
libcaf_net/caf/net/socket_manager.hpp
+1
-1
libcaf_net/src/tcp_accept_socket.cpp
libcaf_net/src/tcp_accept_socket.cpp
+1
-1
libcaf_net/src/tcp_stream_socket.cpp
libcaf_net/src/tcp_stream_socket.cpp
+1
-1
libcaf_net/test/doorman.cpp
libcaf_net/test/doorman.cpp
+42
-30
No files found.
libcaf_net/CMakeLists.txt
View file @
412be4ab
...
...
@@ -36,6 +36,8 @@ set(LIBCAF_NET_SRCS
src/pipe_socket.cpp
src/pollset_updater.cpp
src/socket.cpp
src/tcp_accept_socket.cpp
src/tcp_stream_socket.cpp
src/socket_manager.cpp
src/stream_socket.cpp
src/udp_datagram_socket.cpp
...
...
libcaf_net/caf/net/doorman.hpp
View file @
412be4ab
...
...
@@ -21,34 +21,47 @@
#include <vector>
#include "caf/logger.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/socket.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/send.hpp"
#include "caf/stream_transport.hpp"
namespace
caf
{
namespace
caf
::
net
{
/// A doorman accepts TCP connections and creates stream_transports to handle
/// them.
template
<
class
Factory
>
class
doorman
{
public:
doorman
(
net
::
tcp_accept_socket
acceptor
)
:
acceptor_
(
acceptor
)
{
// -- member types -----------------------------------------------------------
using
factory_type
=
Factory
;
using
application_type
=
typename
Factory
::
application_type
;
// -- constructors, destructors, and assignment operators --------------------
explicit
doorman
(
net
::
tcp_accept_socket
acceptor
,
factory_type
factory
)
:
acceptor_
(
acceptor
),
factory_
(
std
::
move
(
factory
))
{
// nop
}
net
::
tcp_accept_socket
acceptor_
;
// -- properties -------------------------------------------------------------
net
::
tcp_accept_socket
handle
()
{
return
acceptor_
;
}
// -- member functions -------------------------------------------------------
template
<
class
Parent
>
error
init
(
Parent
&
parent
)
{
if
(
auto
err
=
parent
.
application
().
init
(
parent
))
// TODO: is initializing application factory nessecary?
if
(
auto
err
=
factory_
.
init
(
parent
))
return
err
;
parent
.
mask_add
(
net
::
operation
::
read
);
return
none
;
}
...
...
@@ -59,13 +72,16 @@ public:
CAF_LOG_ERROR
(
"accept failed:"
<<
parent
.
system
().
render
(
x
.
error
()));
return
false
;
}
// TODO this won't work anymore
auto
mpx
=
parent
.
multiplexer
();
if
(
!
mpx
)
{
CAF_LOG_DEBUG
(
"unable to get multiplexer from parent"
);
return
false
;
}
auto
child
=
make_endpoint_manager
(
mpx
,
parent
.
system
(),
scribe
{
*
x
},
parent
.
application
().
make
());
auto
child
=
make_endpoint_manager
(
mpx
,
parent
.
system
(),
stream_transport
<
application_type
>
{
*
x
,
factory_
.
make
()});
if
(
auto
err
=
child
->
init
())
return
false
;
return
true
;
...
...
@@ -78,11 +94,26 @@ public:
}
template
<
class
Parent
>
void
resolve
(
Parent
&
,
const
std
::
string
&
path
,
actor
listener
)
{
CAF_LOG_ERROR
(
"doorman called to resolve"
<<
CAF_ARG
(
path
));
void
resolve
(
Parent
&
,
const
uri
&
locator
,
const
actor
&
listener
)
{
CAF_LOG_ERROR
(
"doorman called to resolve"
<<
CAF_ARG
(
locator
));
anon_send
(
listener
,
resolve_atom
::
value
,
"doormen cannot resolve paths"
);
}
void
new_proxy
(
endpoint_manager
&
,
const
node_id
&
peer
,
actor_id
id
)
{
CAF_LOG_ERROR
(
"doorman received new_proxy"
<<
CAF_ARG
(
peer
)
<<
CAF_ARG
(
id
));
CAF_IGNORE_UNUSED
(
peer
);
CAF_IGNORE_UNUSED
(
id
);
}
void
local_actor_down
(
endpoint_manager
&
,
const
node_id
&
peer
,
actor_id
id
,
error
reason
)
{
CAF_LOG_ERROR
(
"doorman received local_actor_down"
<<
CAF_ARG
(
peer
)
<<
CAF_ARG
(
id
)
<<
CAF_ARG
(
reason
));
CAF_IGNORE_UNUSED
(
peer
);
CAF_IGNORE_UNUSED
(
id
);
CAF_IGNORE_UNUSED
(
reason
);
}
template
<
class
Parent
>
void
timeout
(
Parent
&
,
atom_value
x
,
uint64_t
id
)
{
CAF_LOG_ERROR
(
"doorman received timeout"
<<
CAF_ARG
(
x
)
<<
CAF_ARG
(
id
));
...
...
@@ -90,10 +121,14 @@ public:
CAF_IGNORE_UNUSED
(
id
);
}
template
<
class
Application
>
void
handle_error
(
Application
&
,
sec
err
)
{
void
handle_error
(
sec
err
)
{
CAF_LOG_ERROR
(
"doorman encounterd error: "
<<
err
);
}
private:
net
::
tcp_accept_socket
acceptor_
;
factory_type
factory_
;
};
}
// namespace caf
}
// namespace caf
::net
libcaf_net/caf/net/socket_manager.hpp
View file @
412be4ab
...
...
@@ -36,7 +36,7 @@ public:
/// @pre `handle != invalid_socket`
socket_manager
(
socket
handle
,
const
multiplexer_ptr
&
parent
);
virtual
~
socket_manager
()
;
~
socket_manager
()
override
;
socket_manager
(
const
socket_manager
&
)
=
delete
;
...
...
libcaf_net/src/tcp_accept_socket.cpp
View file @
412be4ab
...
...
@@ -118,7 +118,7 @@ make_tcp_accept_socket(const uri::authority_type& node, bool reuse_addr) {
if
(
auto
ip
=
get_if
<
ip_address
>
(
&
node
.
host
))
return
make_tcp_accept_socket
(
ip_endpoint
{
*
ip
,
node
.
port
},
reuse_addr
);
auto
host
=
get
<
std
::
string
>
(
node
.
host
);
auto
addrs
=
ip
::
resolve
(
host
);
auto
addrs
=
ip
::
resolve
(
host
);
// TODO: switch to local_addresses()
if
(
addrs
.
empty
())
return
make_error
(
sec
::
cannot_open_port
,
"no local interface available"
,
to_string
(
node
));
...
...
libcaf_net/src/tcp_stream_socket.cpp
View file @
412be4ab
...
...
@@ -83,7 +83,7 @@ make_connected_tcp_stream_socket(const uri::authority_type& node) {
return
make_error
(
sec
::
cannot_connect_to_node
,
"port is zero"
);
std
::
vector
<
ip_address
>
addrs
;
if
(
auto
str
=
get_if
<
std
::
string
>
(
&
node
.
host
))
addrs
=
ip
::
resolve
(
std
::
move
(
*
str
)
);
addrs
=
ip
::
local_addresses
(
*
str
);
else
if
(
auto
addr
=
get_if
<
ip_address
>
(
&
node
.
host
))
addrs
.
push_back
(
*
addr
);
if
(
addrs
.
empty
())
...
...
libcaf_net/test/doorman.cpp
View file @
412be4ab
...
...
@@ -18,7 +18,7 @@
#define CAF_SUITE doorman
#include "caf/
policy
/doorman.hpp"
#include "caf/
net
/doorman.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/ip.hpp"
...
...
@@ -30,29 +30,26 @@
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include "
caf/net/test/
host_fixture.hpp"
using
namespace
caf
;
using
namespace
caf
::
net
;
using
namespace
std
::
literals
::
string_literals
;
namespace
{
// TODO: switch to std::operator""s when switching to C++14
std
::
string
operator
""
_s
(
const
char
*
str
,
size_t
size
)
{
return
std
::
string
(
str
,
size
);
}
struct
fixture
:
test_coordinator_fixture
<>
,
host_fixture
{
fixture
()
{
mpx
=
std
::
make_shared
<
multiplexer
>
();
if
(
auto
err
=
mpx
->
init
())
CAF_FAIL
(
"mpx->init failed: "
<<
sys
.
render
(
err
));
mpx
->
set_thread_id
();
CAF_CHECK_EQUAL
(
mpx
->
num_socket_managers
(),
1u
);
auth
.
port
=
0
;
auth
.
host
=
"0.0.0.0"
_
s
;
auth
.
host
=
"0.0.0.0"
s
;
}
bool
handle_io_event
()
override
{
mpx
->
handle_updates
();
return
mpx
->
poll_once
(
false
);
}
...
...
@@ -71,30 +68,41 @@ public:
return
result
;
}
template
<
class
Transpor
t
>
error
init
(
Transpor
t
&
)
{
template
<
class
Paren
t
>
error
init
(
Paren
t
&
)
{
return
none
;
}
template
<
class
Transpor
t
>
void
write_message
(
Transport
&
transpor
t
,
std
::
unique_ptr
<
endpoint_manager
::
message
>
msg
)
{
transpor
t
.
write_packet
(
msg
->
payload
);
template
<
class
Paren
t
>
void
write_message
(
Parent
&
paren
t
,
std
::
unique_ptr
<
endpoint_manager
_queue
::
message
>
msg
)
{
paren
t
.
write_packet
(
msg
->
payload
);
}
template
<
class
Parent
>
void
handle_data
(
Parent
&
,
span
<
byte
>
)
{
// nop
error
handle_data
(
Parent
&
,
span
<
const
byte
>
)
{
return
none
;
}
template
<
class
Transpor
t
>
void
resolve
(
Transport
&
,
std
::
string
path
,
actor
listener
)
{
template
<
class
Paren
t
>
void
resolve
(
Parent
&
,
string_view
path
,
const
actor
&
listener
)
{
anon_send
(
listener
,
resolve_atom
::
value
,
"the resolved path is still "
+
path
);
"the resolved path is still "
+
std
::
string
(
path
.
begin
(),
path
.
end
()));
}
template
<
class
Parent
>
void
timeout
(
Parent
&
,
atom_value
,
uint64_t
)
{
// nop
}
template
<
class
Transport
>
void
timeout
(
Transport
&
,
atom_value
,
uint64_t
)
{
template
<
class
Parent
>
void
new_proxy
(
Parent
&
,
actor_id
)
{
// nop
}
template
<
class
Parent
>
void
local_actor_down
(
Parent
&
,
actor_id
,
error
)
{
// nop
}
...
...
@@ -105,6 +113,8 @@ public:
class
dummy_application_factory
{
public:
using
application_type
=
dummy_application
;
static
expected
<
std
::
vector
<
byte
>>
serialize
(
actor_system
&
sys
,
const
type_erased_tuple
&
x
)
{
return
dummy_application
::
serialize
(
sys
,
x
);
...
...
@@ -115,7 +125,7 @@ public:
return
none
;
}
dummy_application
make
()
const
{
application_type
make
()
const
{
return
dummy_application
{};
}
};
...
...
@@ -131,7 +141,7 @@ CAF_TEST(tcp connect) {
CAF_MESSAGE
(
"opened acceptor on port "
<<
port
);
uri
::
authority_type
dst
;
dst
.
port
=
port
;
dst
.
host
=
"localhost"
_
s
;
dst
.
host
=
"localhost"
s
;
auto
conn
=
make_socket_guard
(
unbox
(
make_connected_tcp_stream_socket
(
dst
)));
auto
accepted
=
make_socket_guard
(
unbox
(
accept
(
acceptor
)));
CAF_MESSAGE
(
"accepted connection"
);
...
...
@@ -142,19 +152,21 @@ CAF_TEST(doorman accept) {
auto
port
=
unbox
(
local_port
(
socket_cast
<
network_socket
>
(
acceptor
)));
auto
acceptor_guard
=
make_socket_guard
(
acceptor
);
CAF_MESSAGE
(
"opened acceptor on port "
<<
port
);
auto
mgr
=
make_endpoint_manager
(
mpx
,
sys
,
policy
::
doorman
{
acceptor
},
dummy_application_factory
{});
auto
mgr
=
make_endpoint_manager
(
mpx
,
sys
,
doorman
<
dummy_application_factory
>
{
acceptor_guard
.
release
(),
dummy_application_factory
{}});
CAF_CHECK_EQUAL
(
mgr
->
init
(),
none
);
handle_io_event
();
auto
before
=
mpx
->
num_socket_managers
();
CAF_
MESSAGE
(
"connecting to doorman"
);
CAF_
CHECK_EQUAL
(
before
,
2
);
uri
::
authority_type
dst
;
dst
.
port
=
port
;
dst
.
host
=
"localhost"
_s
;
dst
.
host
=
"localhost"
s
;
CAF_MESSAGE
(
"connecting to doorman on: "
<<
dst
);
auto
conn
=
make_socket_guard
(
unbox
(
make_connected_tcp_stream_socket
(
dst
)));
CAF_MESSAGE
(
"waiting for connection"
);
while
(
mpx
->
num_socket_managers
()
!=
before
+
1
)
handle_io_event
();
run
();
CAF_MESSAGE
(
"connected"
);
}
...
...
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