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
6bfe920c
Commit
6bfe920c
authored
Jun 16, 2020
by
Dominik Charousset
Committed by
GitHub
Jun 16, 2020
Browse files
Options
Browse Files
Download
Plain Diff
Merge pull request #71
Add tcp_backend
parents
cd226607
98dc2020
Changes
19
Show whitespace changes
Inline
Side-by-side
Showing
19 changed files
with
612 additions
and
13 deletions
+612
-13
libcaf_net/CMakeLists.txt
libcaf_net/CMakeLists.txt
+3
-1
libcaf_net/caf/net/backend/tcp.hpp
libcaf_net/caf/net/backend/tcp.hpp
+112
-0
libcaf_net/caf/net/backend/test.hpp
libcaf_net/caf/net/backend/test.hpp
+4
-0
libcaf_net/caf/net/basp/application.hpp
libcaf_net/caf/net/basp/application.hpp
+2
-1
libcaf_net/caf/net/basp/application_factory.hpp
libcaf_net/caf/net/basp/application_factory.hpp
+51
-0
libcaf_net/caf/net/basp/header.hpp
libcaf_net/caf/net/basp/header.hpp
+4
-3
libcaf_net/caf/net/basp/worker.hpp
libcaf_net/caf/net/basp/worker.hpp
+3
-2
libcaf_net/caf/net/defaults.hpp
libcaf_net/caf/net/defaults.hpp
+8
-2
libcaf_net/caf/net/endpoint_manager_queue.hpp
libcaf_net/caf/net/endpoint_manager_queue.hpp
+2
-1
libcaf_net/caf/net/middleman.hpp
libcaf_net/caf/net/middleman.hpp
+42
-0
libcaf_net/caf/net/middleman_backend.hpp
libcaf_net/caf/net/middleman_backend.hpp
+5
-0
libcaf_net/caf/net/multiplexer.hpp
libcaf_net/caf/net/multiplexer.hpp
+3
-1
libcaf_net/caf/net/socket_manager.hpp
libcaf_net/caf/net/socket_manager.hpp
+2
-1
libcaf_net/src/basp/application.cpp
libcaf_net/src/basp/application.cpp
+0
-1
libcaf_net/src/defaults.cpp
libcaf_net/src/defaults.cpp
+2
-0
libcaf_net/src/net/backend/tcp.cpp
libcaf_net/src/net/backend/tcp.cpp
+142
-0
libcaf_net/src/net/backend/test.cpp
libcaf_net/src/net/backend/test.cpp
+12
-0
libcaf_net/src/net/middleman.cpp
libcaf_net/src/net/middleman.cpp
+17
-0
libcaf_net/test/net/backend/tcp.cpp
libcaf_net/test/net/backend/tcp.cpp
+198
-0
No files found.
libcaf_net/CMakeLists.txt
View file @
6bfe920c
...
...
@@ -35,7 +35,7 @@ endfunction()
add_library
(
libcaf_net_obj OBJECT
${
CAF_NET_HEADERS
}
src/actor_proxy_impl.cpp
src/application.cpp
src/
basp/
application.cpp
src/basp/connection_state_strings.cpp
src/basp/ec_strings.cpp
src/basp/message_type_strings.cpp
...
...
@@ -51,6 +51,7 @@ add_library(libcaf_net_obj OBJECT ${CAF_NET_HEADERS}
src/message_queue.cpp
src/multiplexer.cpp
src/net/backend/test.cpp
src/net/backend/tcp.cpp
src/net/endpoint_manager_queue.cpp
src/net/middleman.cpp
src/net/middleman_backend.cpp
...
...
@@ -147,4 +148,5 @@ caf_incubator_add_test_suites(caf-net-test
transport_worker_dispatcher
udp_datagram_socket
network_socket
net.backend.tcp
)
libcaf_net/caf/net/backend/tcp.hpp
0 → 100644
View file @
6bfe920c
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 <map>
#include <mutex>
#include "caf/detail/net_export.hpp"
#include "caf/error.hpp"
#include "caf/expected.hpp"
#include "caf/net/basp/application.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/middleman_backend.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/node_id.hpp"
namespace
caf
::
net
::
backend
{
/// Minimal backend for tcp communication.
class
CAF_NET_EXPORT
tcp
:
public
middleman_backend
{
public:
using
peer_map
=
std
::
map
<
node_id
,
endpoint_manager_ptr
>
;
using
emplace_return_type
=
std
::
pair
<
peer_map
::
iterator
,
bool
>
;
// -- constructors, destructors, and assignment operators --------------------
tcp
(
middleman
&
mm
);
~
tcp
()
override
;
// -- interface functions ----------------------------------------------------
error
init
()
override
;
void
stop
()
override
;
expected
<
endpoint_manager_ptr
>
get_or_connect
(
const
uri
&
locator
)
override
;
endpoint_manager_ptr
peer
(
const
node_id
&
id
)
override
;
void
resolve
(
const
uri
&
locator
,
const
actor
&
listener
)
override
;
strong_actor_ptr
make_proxy
(
node_id
nid
,
actor_id
aid
)
override
;
void
set_last_hop
(
node_id
*
)
override
;
// -- properties -------------------------------------------------------------
uint16_t
port
()
const
noexcept
override
;
template
<
class
Handle
>
expected
<
endpoint_manager_ptr
>
emplace
(
const
node_id
&
peer_id
,
Handle
socket_handle
)
{
using
transport_type
=
stream_transport
<
basp
::
application
>
;
if
(
auto
err
=
nonblocking
(
socket_handle
,
true
))
return
err
;
auto
mpx
=
mm_
.
mpx
();
basp
::
application
app
{
proxies_
};
auto
mgr
=
make_endpoint_manager
(
mpx
,
mm_
.
system
(),
transport_type
{
socket_handle
,
std
::
move
(
app
)});
if
(
auto
err
=
mgr
->
init
())
{
CAF_LOG_ERROR
(
"mgr->init() failed: "
<<
err
);
return
err
;
}
mpx
->
register_reading
(
mgr
);
emplace_return_type
res
;
{
const
std
::
lock_guard
<
std
::
mutex
>
lock
(
lock_
);
res
=
peers_
.
emplace
(
peer_id
,
std
::
move
(
mgr
));
}
if
(
res
.
second
)
return
res
.
first
->
second
;
else
return
make_error
(
sec
::
runtime_error
,
"peer_id already exists"
);
}
private:
endpoint_manager_ptr
get_peer
(
const
node_id
&
id
);
middleman
&
mm_
;
peer_map
peers_
;
proxy_registry
proxies_
;
uint16_t
listening_port_
;
std
::
mutex
lock_
;
};
}
// namespace caf::net::backend
libcaf_net/caf/net/backend/test.hpp
View file @
6bfe920c
...
...
@@ -51,6 +51,8 @@ public:
endpoint_manager_ptr
peer
(
const
node_id
&
id
)
override
;
expected
<
endpoint_manager_ptr
>
get_or_connect
(
const
uri
&
locator
)
override
;
void
resolve
(
const
uri
&
locator
,
const
actor
&
listener
)
override
;
strong_actor_ptr
make_proxy
(
node_id
nid
,
actor_id
aid
)
override
;
...
...
@@ -63,6 +65,8 @@ public:
return
get_peer
(
peer_id
).
first
;
}
uint16_t
port
()
const
noexcept
override
;
peer_entry
&
emplace
(
const
node_id
&
peer_id
,
stream_socket
first
,
stream_socket
second
);
...
...
libcaf_net/caf/net/basp/application.hpp
View file @
6bfe920c
...
...
@@ -31,6 +31,7 @@
#include "caf/byte.hpp"
#include "caf/callback.hpp"
#include "caf/defaults.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/detail/worker_hub.hpp"
#include "caf/error.hpp"
#include "caf/fwd.hpp"
...
...
@@ -52,7 +53,7 @@
namespace
caf
::
net
::
basp
{
/// An implementation of BASP as an application layer protocol.
class
application
{
class
CAF_NET_EXPORT
application
{
public:
// -- member types -----------------------------------------------------------
...
...
libcaf_net/caf/net/basp/application_factory.hpp
0 → 100644
View file @
6bfe920c
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/detail/net_export.hpp"
#include "caf/error.hpp"
#include "caf/net/basp/application.hpp"
#include "caf/proxy_registry.hpp"
namespace
caf
::
net
::
basp
{
/// Factory for basp::application.
/// @relates doorman
class
CAF_NET_EXPORT
application_factory
{
public:
using
application_type
=
basp
::
application
;
application_factory
(
proxy_registry
&
proxies
)
:
proxies_
(
proxies
)
{
// nop
}
template
<
class
Parent
>
error
init
(
Parent
&
)
{
return
none
;
}
application_type
make
()
const
{
return
application_type
{
proxies_
};
}
private:
proxy_registry
&
proxies_
;
};
}
// namespace caf::net::basp
libcaf_net/caf/net/basp/header.hpp
View file @
6bfe920c
...
...
@@ -22,6 +22,7 @@
#include <cstdint>
#include "caf/detail/comparable.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include "caf/meta/hex_formatted.hpp"
#include "caf/meta/type_name.hpp"
...
...
@@ -33,7 +34,7 @@ namespace caf::net::basp {
/// @addtogroup BASP
/// The header of a Binary Actor System Protocol (BASP) message.
struct
header
:
detail
::
comparable
<
header
>
{
struct
CAF_NET_EXPORT
header
:
detail
::
comparable
<
header
>
{
// -- constructors, destructors, and assignment operators --------------------
constexpr
header
()
noexcept
...
...
@@ -74,11 +75,11 @@ struct header : detail::comparable<header> {
/// Serializes a header to a byte representation.
/// @relates header
std
::
array
<
byte
,
header_size
>
to_bytes
(
header
x
);
CAF_NET_EXPORT
std
::
array
<
byte
,
header_size
>
to_bytes
(
header
x
);
/// Serializes a header to a byte representation.
/// @relates header
void
to_bytes
(
header
x
,
byte_buffer
&
buf
);
CAF_NET_EXPORT
void
to_bytes
(
header
x
,
byte_buffer
&
buf
);
/// @relates header
template
<
class
Inspector
>
...
...
libcaf_net/caf/net/basp/worker.hpp
View file @
6bfe920c
...
...
@@ -24,6 +24,7 @@
#include "caf/byte_buffer.hpp"
#include "caf/config.hpp"
#include "caf/detail/abstract_worker.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/detail/worker_hub.hpp"
#include "caf/fwd.hpp"
#include "caf/net/basp/header.hpp"
...
...
@@ -36,7 +37,7 @@
namespace
caf
::
net
::
basp
{
/// Deserializes payloads for BASP messages asynchronously.
class
worker
:
public
detail
::
abstract_worker
,
class
CAF_NET_EXPORT
worker
:
public
detail
::
abstract_worker
,
public
remote_message_handler
<
worker
>
{
public:
// -- friends ----------------------------------------------------------------
...
...
libcaf_net/caf/net/defaults.hpp
View file @
6bfe920c
...
...
@@ -19,15 +19,21 @@
#pragma once
#include <cstddef>
#include <cstdint>
#include "caf/detail/net_export.hpp"
// -- hard-coded default values for various CAF options ------------------------
namespace
caf
::
defaults
::
middleman
{
/// Maximum number of cached buffers for sending payloads.
extern
const
size_t
max_payload_buffers
;
CAF_NET_EXPORT
extern
const
size_t
max_payload_buffers
;
/// Maximum number of cached buffers for sending headers.
extern
const
size_t
max_header_buffers
;
CAF_NET_EXPORT
extern
const
size_t
max_header_buffers
;
/// Port to listen on for tcp.
CAF_NET_EXPORT
extern
const
uint16_t
tcp_port
;
}
// namespace caf::defaults::middleman
libcaf_net/caf/net/endpoint_manager_queue.hpp
View file @
6bfe920c
...
...
@@ -21,6 +21,7 @@
#include <string>
#include "caf/actor.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/detail/serialized_size.hpp"
#include "caf/fwd.hpp"
#include "caf/intrusive/drr_queue.hpp"
...
...
@@ -33,7 +34,7 @@
namespace
caf
::
net
{
class
endpoint_manager_queue
{
class
CAF_NET_EXPORT
endpoint_manager_queue
{
public:
enum
class
element_type
{
event
,
message
};
...
...
libcaf_net/caf/net/middleman.hpp
View file @
6bfe920c
...
...
@@ -18,6 +18,9 @@
#pragma once
#include <chrono>
#include <set>
#include <string>
#include <thread>
#include "caf/actor_system.hpp"
...
...
@@ -25,6 +28,8 @@
#include "caf/detail/type_list.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/middleman_backend.hpp"
#include "caf/scoped_actor.hpp"
namespace
caf
::
net
{
...
...
@@ -72,9 +77,44 @@ public:
// -- remoting ---------------------------------------------------------------
expected
<
endpoint_manager_ptr
>
connect
(
const
uri
&
locator
);
// Publishes an actor.
template
<
class
Handle
>
void
publish
(
Handle
whom
,
const
std
::
string
&
path
)
{
// TODO: Currently, we can't get the interface from the registry. Either we
// change that, or we need to associate the handle with the interface.
system
().
registry
().
put
(
path
,
whom
);
}
/// Resolves a path to a remote actor.
void
resolve
(
const
uri
&
locator
,
const
actor
&
listener
);
template
<
class
Handle
=
actor
,
class
Duration
=
std
::
chrono
::
seconds
>
expected
<
Handle
>
remote_actor
(
const
uri
&
locator
,
Duration
timeout
=
std
::
chrono
::
seconds
(
5
))
{
scoped_actor
self
{
sys_
};
resolve
(
locator
,
self
);
Handle
handle
;
error
err
;
self
->
receive
(
[
&
handle
](
strong_actor_ptr
&
ptr
,
const
std
::
set
<
std
::
string
>&
)
{
// TODO: This cast is not type-safe.
handle
=
actor_cast
<
Handle
>
(
std
::
move
(
ptr
));
},
[
&
err
](
const
error
&
e
)
{
err
=
e
;
},
after
(
timeout
)
>>
[
&
err
]
{
err
=
make_error
(
sec
::
runtime_error
,
"manager did not respond with a proxy."
);
});
if
(
err
)
return
err
;
if
(
handle
)
return
handle
;
return
make_error
(
sec
::
runtime_error
,
"cast to handle-type failed"
);
}
// -- properties -------------------------------------------------------------
actor_system
&
system
()
{
...
...
@@ -91,6 +131,8 @@ public:
middleman_backend
*
backend
(
string_view
scheme
)
const
noexcept
;
expected
<
uint16_t
>
port
(
string_view
scheme
)
const
;
private:
// -- constructors, destructors, and assignment operators --------------------
...
...
libcaf_net/caf/net/middleman_backend.hpp
View file @
6bfe920c
...
...
@@ -46,6 +46,9 @@ public:
/// @returns The endpoint manager for `peer` on success, `nullptr` otherwise.
virtual
endpoint_manager_ptr
peer
(
const
node_id
&
id
)
=
0
;
/// Establishes a connection to a remote node.
virtual
expected
<
endpoint_manager_ptr
>
get_or_connect
(
const
uri
&
locator
)
=
0
;
/// Resolves a path to a remote actor.
virtual
void
resolve
(
const
uri
&
locator
,
const
actor
&
listener
)
=
0
;
...
...
@@ -57,6 +60,8 @@ public:
return
id_
;
}
virtual
uint16_t
port
()
const
=
0
;
private:
/// Stores the technology-specific identifier.
std
::
string
id_
;
...
...
libcaf_net/caf/net/multiplexer.hpp
View file @
6bfe920c
...
...
@@ -22,6 +22,7 @@
#include <mutex>
#include <thread>
#include "caf/detail/net_export.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/operation.hpp"
#include "caf/net/pipe_socket.hpp"
...
...
@@ -37,7 +38,8 @@ struct pollfd;
namespace
caf
::
net
{
/// Multiplexes any number of ::socket_manager objects with a ::socket.
class
multiplexer
:
public
std
::
enable_shared_from_this
<
multiplexer
>
{
class
CAF_NET_EXPORT
multiplexer
:
public
std
::
enable_shared_from_this
<
multiplexer
>
{
public:
// -- member types -----------------------------------------------------------
...
...
libcaf_net/caf/net/socket_manager.hpp
View file @
6bfe920c
...
...
@@ -18,6 +18,7 @@
#pragma once
#include "caf/detail/net_export.hpp"
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
...
...
@@ -28,7 +29,7 @@
namespace
caf
::
net
{
/// Manages the lifetime of a single socket and handles any I/O events on it.
class
socket_manager
:
public
ref_counted
{
class
CAF_NET_EXPORT
socket_manager
:
public
ref_counted
{
public:
// -- constructors, destructors, and assignment operators --------------------
...
...
libcaf_net/src/application.cpp
→
libcaf_net/src/
basp/
application.cpp
View file @
6bfe920c
...
...
@@ -29,7 +29,6 @@
#include "caf/detail/network_order.hpp"
#include "caf/detail/parse.hpp"
#include "caf/error.hpp"
#include "caf/expected.hpp"
#include "caf/logger.hpp"
#include "caf/net/basp/constants.hpp"
#include "caf/net/basp/ec.hpp"
...
...
libcaf_net/src/defaults.cpp
View file @
6bfe920c
...
...
@@ -24,4 +24,6 @@ const size_t max_payload_buffers = 100;
const
size_t
max_header_buffers
=
10
;
const
uint16_t
tcp_port
=
0
;
}
// namespace caf::defaults::middleman
libcaf_net/src/net/backend/tcp.cpp
0 → 100644
View file @
6bfe920c
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/backend/tcp.hpp"
#include <mutex>
#include <string>
#include "caf/net/actor_proxy_impl.hpp"
#include "caf/net/basp/application.hpp"
#include "caf/net/basp/application_factory.hpp"
#include "caf/net/basp/ec.hpp"
#include "caf/net/doorman.hpp"
#include "caf/net/ip.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/socket_guard.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
::
backend
{
tcp
::
tcp
(
middleman
&
mm
)
:
middleman_backend
(
"tcp"
),
mm_
(
mm
),
proxies_
(
mm
.
system
(),
*
this
)
{
// nop
}
tcp
::~
tcp
()
{
// nop
}
error
tcp
::
init
()
{
uint16_t
conf_port
=
get_or
<
uint16_t
>
(
mm_
.
system
().
config
(),
"middleman.tcp-port"
,
defaults
::
middleman
::
tcp_port
);
ip_endpoint
ep
;
auto
local_address
=
std
::
string
(
"[::]:"
)
+
std
::
to_string
(
conf_port
);
if
(
auto
err
=
detail
::
parse
(
local_address
,
ep
))
return
err
;
auto
acceptor
=
make_tcp_accept_socket
(
ep
,
true
);
if
(
!
acceptor
)
return
acceptor
.
error
();
auto
acc_guard
=
make_socket_guard
(
*
acceptor
);
if
(
auto
err
=
nonblocking
(
acc_guard
.
socket
(),
true
))
return
err
;
auto
port
=
local_port
(
*
acceptor
);
if
(
!
port
)
return
port
.
error
();
listening_port_
=
*
port
;
CAF_LOG_INFO
(
"doorman spawned on "
<<
CAF_ARG
(
*
port
));
auto
doorman_uri
=
make_uri
(
"tcp://doorman"
);
if
(
!
doorman_uri
)
return
doorman_uri
.
error
();
auto
&
mpx
=
mm_
.
mpx
();
auto
mgr
=
make_endpoint_manager
(
mpx
,
mm_
.
system
(),
doorman
{
acc_guard
.
release
(),
basp
::
application_factory
{
proxies_
}});
if
(
auto
err
=
mgr
->
init
())
{
CAF_LOG_ERROR
(
"mgr->init() failed: "
<<
err
);
return
err
;
}
return
none
;
}
void
tcp
::
stop
()
{
for
(
const
auto
&
p
:
peers_
)
proxies_
.
erase
(
p
.
first
);
peers_
.
clear
();
}
expected
<
endpoint_manager_ptr
>
tcp
::
get_or_connect
(
const
uri
&
locator
)
{
if
(
auto
auth
=
locator
.
authority_only
())
{
auto
id
=
make_node_id
(
*
auth
);
if
(
auto
ptr
=
peer
(
id
))
return
ptr
;
auto
host
=
locator
.
authority
().
host
;
if
(
auto
hostname
=
get_if
<
std
::
string
>
(
&
host
))
{
for
(
const
auto
&
addr
:
ip
::
resolve
(
*
hostname
))
{
ip_endpoint
ep
{
addr
,
locator
.
authority
().
port
};
auto
sock
=
make_connected_tcp_stream_socket
(
ep
);
if
(
!
sock
)
continue
;
else
return
emplace
(
id
,
*
sock
);
}
}
}
return
sec
::
cannot_connect_to_node
;
}
endpoint_manager_ptr
tcp
::
peer
(
const
node_id
&
id
)
{
return
get_peer
(
id
);
}
void
tcp
::
resolve
(
const
uri
&
locator
,
const
actor
&
listener
)
{
if
(
auto
p
=
get_or_connect
(
locator
))
(
*
p
)
->
resolve
(
locator
,
listener
);
else
anon_send
(
listener
,
p
.
error
());
}
strong_actor_ptr
tcp
::
make_proxy
(
node_id
nid
,
actor_id
aid
)
{
using
impl_type
=
actor_proxy_impl
;
using
hdl_type
=
strong_actor_ptr
;
actor_config
cfg
;
return
make_actor
<
impl_type
,
hdl_type
>
(
aid
,
nid
,
&
mm_
.
system
(),
cfg
,
peer
(
nid
));
}
void
tcp
::
set_last_hop
(
node_id
*
)
{
// nop
}
uint16_t
tcp
::
port
()
const
noexcept
{
return
listening_port_
;
}
endpoint_manager_ptr
tcp
::
get_peer
(
const
node_id
&
id
)
{
const
std
::
lock_guard
<
std
::
mutex
>
lock
(
lock_
);
auto
i
=
peers_
.
find
(
id
);
if
(
i
!=
peers_
.
end
())
return
i
->
second
;
return
nullptr
;
}
}
// namespace caf::net::backend
libcaf_net/src/net/backend/test.cpp
View file @
6bfe920c
...
...
@@ -27,6 +27,7 @@
#include "caf/net/multiplexer.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/raise_error.hpp"
#include "caf/sec.hpp"
#include "caf/send.hpp"
namespace
caf
::
net
::
backend
{
...
...
@@ -54,6 +55,13 @@ endpoint_manager_ptr test::peer(const node_id& id) {
return
get_peer
(
id
).
second
;
}
expected
<
endpoint_manager_ptr
>
test
::
get_or_connect
(
const
uri
&
locator
)
{
if
(
auto
ptr
=
peer
(
make_node_id
(
*
locator
.
authority_only
())))
return
ptr
;
return
make_error
(
sec
::
runtime_error
,
"connecting not implemented in test backend"
);
}
void
test
::
resolve
(
const
uri
&
locator
,
const
actor
&
listener
)
{
auto
id
=
locator
.
authority_only
();
if
(
id
)
...
...
@@ -74,6 +82,10 @@ void test::set_last_hop(node_id*) {
// nop
}
uint16_t
test
::
port
()
const
noexcept
{
return
0
;
}
test
::
peer_entry
&
test
::
emplace
(
const
node_id
&
peer_id
,
stream_socket
first
,
stream_socket
second
)
{
using
transport_type
=
stream_transport
<
basp
::
application
>
;
...
...
libcaf_net/src/net/middleman.cpp
View file @
6bfe920c
...
...
@@ -20,11 +20,14 @@
#include "caf/actor_system_config.hpp"
#include "caf/detail/set_thread_name.hpp"
#include "caf/expected.hpp"
#include "caf/init_global_meta_objects.hpp"
#include "caf/net/basp/ec.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/middleman_backend.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/raise_error.hpp"
#include "caf/sec.hpp"
#include "caf/send.hpp"
#include "caf/uri.hpp"
...
...
@@ -93,6 +96,13 @@ void* middleman::subtype_ptr() {
return
this
;
}
expected
<
endpoint_manager_ptr
>
middleman
::
connect
(
const
uri
&
locator
)
{
if
(
auto
ptr
=
backend
(
locator
.
scheme
()))
return
ptr
->
get_or_connect
(
locator
);
else
return
basp
::
ec
::
invalid_scheme
;
}
void
middleman
::
resolve
(
const
uri
&
locator
,
const
actor
&
listener
)
{
auto
ptr
=
backend
(
locator
.
scheme
());
if
(
ptr
!=
nullptr
)
...
...
@@ -111,4 +121,11 @@ middleman_backend* middleman::backend(string_view scheme) const noexcept {
return
nullptr
;
}
expected
<
uint16_t
>
middleman
::
port
(
string_view
scheme
)
const
{
if
(
auto
ptr
=
backend
(
scheme
))
return
ptr
->
port
();
else
return
basp
::
ec
::
invalid_scheme
;
}
}
// namespace caf::net
libcaf_net/test/net/backend/tcp.cpp
0 → 100644
View file @
6bfe920c
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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.backend.tcp
#include "caf/net/backend/tcp.hpp"
#include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp"
#include <string>
#include <thread>
#include "caf/actor_system_config.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/socket_guard.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/uri.hpp"
using
namespace
caf
;
using
namespace
caf
::
net
;
using
namespace
std
::
literals
::
string_literals
;
namespace
{
behavior
dummy_actor
(
event_based_actor
*
)
{
return
{
// nop
};
}
struct
earth_node
{
uri
operator
()()
{
return
unbox
(
make_uri
(
"tcp://earth"
));
}
};
struct
mars_node
{
uri
operator
()()
{
return
unbox
(
make_uri
(
"tcp://mars"
));
}
};
template
<
class
Node
>
struct
config
:
actor_system_config
{
config
()
{
Node
this_node
;
put
(
content
,
"middleman.this-node"
,
this_node
());
load
<
middleman
,
backend
::
tcp
>
();
}
};
class
planet_driver
{
public:
virtual
~
planet_driver
()
=
default
;
virtual
bool
handle_io_event
()
=
0
;
};
template
<
class
Node
>
class
planet
:
public
test_coordinator_fixture
<
config
<
Node
>>
{
public:
planet
(
planet_driver
&
driver
)
:
mm
(
this
->
sys
.
network_manager
()),
mpx
(
mm
.
mpx
()),
driver_
(
driver
)
{
mpx
->
set_thread_id
();
}
node_id
id
()
const
{
return
this
->
sys
.
node
();
}
bool
handle_io_event
()
override
{
return
driver_
.
handle_io_event
();
}
net
::
middleman
&
mm
;
multiplexer_ptr
mpx
;
private:
planet_driver
&
driver_
;
};
struct
fixture
:
host_fixture
,
planet_driver
{
fixture
()
:
earth
(
*
this
),
mars
(
*
this
)
{
earth
.
run
();
mars
.
run
();
CAF_REQUIRE_EQUAL
(
earth
.
mpx
->
num_socket_managers
(),
2
);
CAF_REQUIRE_EQUAL
(
mars
.
mpx
->
num_socket_managers
(),
2
);
}
bool
handle_io_event
()
override
{
return
earth
.
mpx
->
poll_once
(
false
)
||
mars
.
mpx
->
poll_once
(
false
);
}
void
run
()
{
earth
.
run
();
}
planet
<
earth_node
>
earth
;
planet
<
mars_node
>
mars
;
};
}
// namespace
CAF_TEST_FIXTURE_SCOPE
(
tcp_backend_tests
,
fixture
)
CAF_TEST
(
doorman
accept
)
{
auto
backend
=
earth
.
mm
.
backend
(
"tcp"
);
CAF_CHECK
(
backend
);
uri
::
authority_type
auth
;
auth
.
host
=
"localhost"
s
;
auth
.
port
=
backend
->
port
();
CAF_MESSAGE
(
"trying to connect to earth at "
<<
CAF_ARG
(
auth
));
auto
sock
=
make_connected_tcp_stream_socket
(
auth
);
CAF_CHECK
(
sock
);
auto
guard
=
make_socket_guard
(
*
sock
);
int
runs
=
0
;
while
(
earth
.
mpx
->
num_socket_managers
()
<
3
)
{
if
(
++
runs
>=
5
)
CAF_FAIL
(
"doorman did not create endpoint_manager"
);
run
();
}
CAF_CHECK_EQUAL
(
earth
.
mpx
->
num_socket_managers
(),
3
);
}
CAF_TEST
(
connect
)
{
uri
::
authority_type
auth
;
auth
.
host
=
"0.0.0.0"
s
;
auth
.
port
=
0
;
auto
acceptor
=
unbox
(
make_tcp_accept_socket
(
auth
,
false
));
auto
acc_guard
=
make_socket_guard
(
acceptor
);
auto
port
=
unbox
(
local_port
(
acc_guard
.
socket
()));
auto
uri_str
=
std
::
string
(
"tcp://localhost:"
)
+
std
::
to_string
(
port
);
CAF_MESSAGE
(
"connecting to "
<<
CAF_ARG
(
uri_str
));
CAF_CHECK
(
earth
.
mm
.
connect
(
*
make_uri
(
uri_str
)));
auto
sock
=
unbox
(
accept
(
acc_guard
.
socket
()));
auto
sock_guard
=
make_socket_guard
(
sock
);
handle_io_event
();
CAF_CHECK_EQUAL
(
earth
.
mpx
->
num_socket_managers
(),
3
);
}
CAF_TEST
(
publish
)
{
auto
dummy
=
earth
.
sys
.
spawn
(
dummy_actor
);
auto
path
=
"dummy"
s
;
CAF_MESSAGE
(
"publishing actor "
<<
CAF_ARG
(
path
));
earth
.
mm
.
publish
(
dummy
,
path
);
CAF_MESSAGE
(
"check registry for "
<<
CAF_ARG
(
path
));
CAF_CHECK_NOT_EQUAL
(
earth
.
sys
.
registry
().
get
(
path
),
nullptr
);
}
CAF_TEST
(
resolve
)
{
using
std
::
chrono
::
milliseconds
;
using
std
::
chrono
::
seconds
;
auto
sockets
=
unbox
(
make_stream_socket_pair
());
auto
earth_be
=
reinterpret_cast
<
net
::
backend
::
tcp
*>
(
earth
.
mm
.
backend
(
"tcp"
));
CAF_CHECK
(
earth_be
->
emplace
(
mars
.
id
(),
sockets
.
first
));
auto
mars_be
=
reinterpret_cast
<
net
::
backend
::
tcp
*>
(
mars
.
mm
.
backend
(
"tcp"
));
CAF_CHECK
(
mars_be
->
emplace
(
earth
.
id
(),
sockets
.
second
));
handle_io_event
();
CAF_CHECK_EQUAL
(
earth
.
mpx
->
num_socket_managers
(),
3
);
CAF_CHECK_EQUAL
(
mars
.
mpx
->
num_socket_managers
(),
3
);
auto
dummy
=
earth
.
sys
.
spawn
(
dummy_actor
);
earth
.
mm
.
publish
(
dummy
,
"dummy"
s
);
auto
locator
=
unbox
(
make_uri
(
"tcp://earth/name/dummy"
s
));
CAF_MESSAGE
(
"resolve "
<<
CAF_ARG
(
locator
));
mars
.
mm
.
resolve
(
locator
,
mars
.
self
);
mars
.
run
();
earth
.
run
();
mars
.
run
();
mars
.
self
->
receive
(
[](
strong_actor_ptr
&
ptr
,
const
std
::
set
<
std
::
string
>&
)
{
CAF_MESSAGE
(
"resolved actor!"
);
CAF_CHECK_NOT_EQUAL
(
ptr
,
nullptr
);
},
[](
const
error
&
err
)
{
CAF_FAIL
(
"got error while resolving: "
<<
CAF_ARG
(
err
));
},
after
(
std
::
chrono
::
seconds
(
0
))
>>
[]
{
CAF_FAIL
(
"manager did not respond with a proxy."
);
});
}
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