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
90ea76ad
Commit
90ea76ad
authored
Dec 02, 2019
by
Dominik Charousset
Committed by
GitHub
Dec 02, 2019
Browse files
Options
Browse Files
Download
Plain Diff
Merge pull request #46
Add Doorman
parents
41d0fca4
8eb90c6a
Changes
9
Hide whitespace changes
Inline
Side-by-side
Showing
9 changed files
with
375 additions
and
6 deletions
+375
-6
libcaf_net/CMakeLists.txt
libcaf_net/CMakeLists.txt
+2
-2
libcaf_net/caf/net/doorman.hpp
libcaf_net/caf/net/doorman.hpp
+133
-0
libcaf_net/caf/net/socket_manager.hpp
libcaf_net/caf/net/socket_manager.hpp
+6
-1
libcaf_net/src/network_socket.cpp
libcaf_net/src/network_socket.cpp
+0
-1
libcaf_net/src/stream_socket.cpp
libcaf_net/src/stream_socket.cpp
+3
-0
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/accept_socket.cpp
libcaf_net/test/accept_socket.cpp
+68
-0
libcaf_net/test/doorman.cpp
libcaf_net/test/doorman.cpp
+161
-0
No files found.
libcaf_net/CMakeLists.txt
View file @
90ea76ad
...
@@ -36,10 +36,10 @@ set(LIBCAF_NET_SRCS
...
@@ -36,10 +36,10 @@ set(LIBCAF_NET_SRCS
src/pipe_socket.cpp
src/pipe_socket.cpp
src/pollset_updater.cpp
src/pollset_updater.cpp
src/socket.cpp
src/socket.cpp
src/socket_manager.cpp
src/stream_socket.cpp
src/tcp_accept_socket.cpp
src/tcp_accept_socket.cpp
src/tcp_stream_socket.cpp
src/tcp_stream_socket.cpp
src/socket_manager.cpp
src/stream_socket.cpp
src/udp_datagram_socket.cpp
src/udp_datagram_socket.cpp
src/defaults.cpp
src/defaults.cpp
src/message_queue.cpp
src/message_queue.cpp
...
...
libcaf_net/caf/net/doorman.hpp
0 → 100644
View file @
90ea76ad
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 <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"
namespace
caf
::
net
{
/// A doorman accepts TCP connections and creates stream_transports to handle
/// them.
template
<
class
Factory
>
class
doorman
{
public:
// -- 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
}
// -- properties -------------------------------------------------------------
net
::
tcp_accept_socket
handle
()
{
return
acceptor_
;
}
// -- member functions -------------------------------------------------------
template
<
class
Parent
>
error
init
(
Parent
&
parent
)
{
// TODO: is initializing application factory nessecary?
if
(
auto
err
=
factory_
.
init
(
parent
))
return
err
;
return
none
;
}
template
<
class
Parent
>
bool
handle_read_event
(
Parent
&
parent
)
{
auto
x
=
net
::
accept
(
acceptor_
);
if
(
!
x
)
{
CAF_LOG_ERROR
(
"accept failed:"
<<
parent
.
system
().
render
(
x
.
error
()));
return
false
;
}
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
(),
stream_transport
<
application_type
>
{
*
x
,
factory_
.
make
()});
if
(
auto
err
=
child
->
init
())
return
false
;
return
true
;
}
template
<
class
Parent
>
bool
handle_write_event
(
Parent
&
)
{
CAF_LOG_ERROR
(
"doorman received write event"
);
return
false
;
}
template
<
class
Parent
>
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
));
CAF_IGNORE_UNUSED
(
x
);
CAF_IGNORE_UNUSED
(
id
);
}
void
handle_error
(
sec
err
)
{
CAF_LOG_ERROR
(
"doorman encounterd error: "
<<
err
);
}
private:
net
::
tcp_accept_socket
acceptor_
;
factory_type
factory_
;
};
}
// namespace caf::net
libcaf_net/caf/net/socket_manager.hpp
View file @
90ea76ad
...
@@ -36,7 +36,7 @@ public:
...
@@ -36,7 +36,7 @@ public:
/// @pre `handle != invalid_socket`
/// @pre `handle != invalid_socket`
socket_manager
(
socket
handle
,
const
multiplexer_ptr
&
parent
);
socket_manager
(
socket
handle
,
const
multiplexer_ptr
&
parent
);
virtual
~
socket_manager
()
;
~
socket_manager
()
override
;
socket_manager
(
const
socket_manager
&
)
=
delete
;
socket_manager
(
const
socket_manager
&
)
=
delete
;
...
@@ -49,6 +49,11 @@ public:
...
@@ -49,6 +49,11 @@ public:
return
handle_
;
return
handle_
;
}
}
/// Returns a pointer to the multiplexer running this `socket_manager`.
multiplexer_ptr
multiplexer
()
const
{
return
parent_
.
lock
();
}
/// Returns registered operations (read, write, or both).
/// Returns registered operations (read, write, or both).
operation
mask
()
const
noexcept
{
operation
mask
()
const
noexcept
{
return
mask_
;
return
mask_
;
...
...
libcaf_net/src/network_socket.cpp
View file @
90ea76ad
...
@@ -26,7 +26,6 @@
...
@@ -26,7 +26,6 @@
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/error.hpp"
#include "caf/error.hpp"
#include "caf/expected.hpp"
#include "caf/expected.hpp"
#include "caf/io/network/protocol.hpp"
#include "caf/logger.hpp"
#include "caf/logger.hpp"
#include "caf/sec.hpp"
#include "caf/sec.hpp"
#include "caf/variant.hpp"
#include "caf/variant.hpp"
...
...
libcaf_net/src/stream_socket.cpp
View file @
90ea76ad
...
@@ -24,6 +24,9 @@
...
@@ -24,6 +24,9 @@
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/expected.hpp"
#include "caf/expected.hpp"
#include "caf/logger.hpp"
#include "caf/logger.hpp"
#include "caf/net/socket.hpp"
#include "caf/net/socket_guard.hpp"
#include "caf/span.hpp"
#include "caf/variant.hpp"
#include "caf/variant.hpp"
#ifdef CAF_POSIX
#ifdef CAF_POSIX
...
...
libcaf_net/src/tcp_accept_socket.cpp
View file @
90ea76ad
...
@@ -118,7 +118,7 @@ make_tcp_accept_socket(const uri::authority_type& node, bool reuse_addr) {
...
@@ -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
))
if
(
auto
ip
=
get_if
<
ip_address
>
(
&
node
.
host
))
return
make_tcp_accept_socket
(
ip_endpoint
{
*
ip
,
node
.
port
},
reuse_addr
);
return
make_tcp_accept_socket
(
ip_endpoint
{
*
ip
,
node
.
port
},
reuse_addr
);
auto
host
=
get
<
std
::
string
>
(
node
.
host
);
auto
host
=
get
<
std
::
string
>
(
node
.
host
);
auto
addrs
=
ip
::
resolve
(
host
);
auto
addrs
=
ip
::
local_addresses
(
host
);
if
(
addrs
.
empty
())
if
(
addrs
.
empty
())
return
make_error
(
sec
::
cannot_open_port
,
"no local interface available"
,
return
make_error
(
sec
::
cannot_open_port
,
"no local interface available"
,
to_string
(
node
));
to_string
(
node
));
...
...
libcaf_net/src/tcp_stream_socket.cpp
View file @
90ea76ad
...
@@ -83,7 +83,7 @@ make_connected_tcp_stream_socket(const uri::authority_type& node) {
...
@@ -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"
);
return
make_error
(
sec
::
cannot_connect_to_node
,
"port is zero"
);
std
::
vector
<
ip_address
>
addrs
;
std
::
vector
<
ip_address
>
addrs
;
if
(
auto
str
=
get_if
<
std
::
string
>
(
&
node
.
host
))
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
))
else
if
(
auto
addr
=
get_if
<
ip_address
>
(
&
node
.
host
))
addrs
.
push_back
(
*
addr
);
addrs
.
push_back
(
*
addr
);
if
(
addrs
.
empty
())
if
(
addrs
.
empty
())
...
...
libcaf_net/test/accept_socket.cpp
0 → 100644
View file @
90ea76ad
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 net.tcp_accept_socket
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/ip.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/socket_guard.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/uri.hpp"
#include "caf/test/dsl.hpp"
#include "caf/net/test/host_fixture.hpp"
using
namespace
caf
;
using
namespace
caf
::
net
;
using
namespace
std
::
literals
::
string_literals
;
namespace
{
struct
fixture
:
test_coordinator_fixture
<>
,
host_fixture
{
fixture
()
{
auth
.
port
=
0
;
auth
.
host
=
"0.0.0.0"
s
;
}
uri
::
authority_type
auth
;
};
}
// namespace
CAF_TEST_FIXTURE_SCOPE
(
doorman_tests
,
fixture
)
CAF_TEST
(
tcp
connect
)
{
auto
acceptor
=
unbox
(
make_tcp_accept_socket
(
auth
,
false
));
auto
port
=
unbox
(
local_port
(
socket_cast
<
network_socket
>
(
acceptor
)));
auto
acceptor_guard
=
make_socket_guard
(
acceptor
);
CAF_MESSAGE
(
"opened acceptor on port "
<<
port
);
uri
::
authority_type
dst
;
dst
.
port
=
port
;
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"
);
}
CAF_TEST_FIXTURE_SCOPE_END
()
libcaf_net/test/doorman.cpp
0 → 100644
View file @
90ea76ad
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 net.doorman
#include "caf/net/doorman.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/ip.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/socket_guard.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/uri.hpp"
#include "caf/test/dsl.hpp"
#include "caf/net/test/host_fixture.hpp"
using
namespace
caf
;
using
namespace
caf
::
net
;
using
namespace
std
::
literals
::
string_literals
;
namespace
{
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
;
}
bool
handle_io_event
()
override
{
return
mpx
->
poll_once
(
false
);
}
multiplexer_ptr
mpx
;
uri
::
authority_type
auth
;
};
class
dummy_application
{
public:
static
expected
<
std
::
vector
<
byte
>>
serialize
(
actor_system
&
sys
,
const
type_erased_tuple
&
x
)
{
std
::
vector
<
byte
>
result
;
binary_serializer
sink
{
sys
,
result
};
if
(
auto
err
=
message
::
save
(
sink
,
x
))
return
err
.
value
();
return
result
;
}
template
<
class
Parent
>
error
init
(
Parent
&
)
{
return
none
;
}
template
<
class
Parent
>
void
write_message
(
Parent
&
parent
,
std
::
unique_ptr
<
endpoint_manager_queue
::
message
>
msg
)
{
parent
.
write_packet
(
msg
->
payload
);
}
template
<
class
Parent
>
error
handle_data
(
Parent
&
,
span
<
const
byte
>
)
{
return
none
;
}
template
<
class
Parent
>
void
resolve
(
Parent
&
,
string_view
path
,
const
actor
&
listener
)
{
anon_send
(
listener
,
resolve_atom
::
value
,
"the resolved path is still "
+
std
::
string
(
path
.
begin
(),
path
.
end
()));
}
template
<
class
Parent
>
void
timeout
(
Parent
&
,
atom_value
,
uint64_t
)
{
// nop
}
template
<
class
Parent
>
void
new_proxy
(
Parent
&
,
actor_id
)
{
// nop
}
template
<
class
Parent
>
void
local_actor_down
(
Parent
&
,
actor_id
,
error
)
{
// nop
}
void
handle_error
(
sec
)
{
// nop
}
};
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
);
}
template
<
class
Parent
>
error
init
(
Parent
&
)
{
return
none
;
}
application_type
make
()
const
{
return
dummy_application
{};
}
};
}
// namespace
CAF_TEST_FIXTURE_SCOPE
(
doorman_tests
,
fixture
)
CAF_TEST
(
doorman
accept
)
{
auto
acceptor
=
unbox
(
make_tcp_accept_socket
(
auth
,
false
));
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
,
doorman
<
dummy_application_factory
>
{
acceptor_guard
.
release
(),
dummy_application_factory
{}});
CAF_CHECK_EQUAL
(
mgr
->
init
(),
none
);
auto
before
=
mpx
->
num_socket_managers
();
CAF_CHECK_EQUAL
(
before
,
2u
);
uri
::
authority_type
dst
;
dst
.
port
=
port
;
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
)
run
();
CAF_MESSAGE
(
"connected"
);
}
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