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
a6183018
Commit
a6183018
authored
Sep 19, 2019
by
Dominik Charousset
Browse files
Options
Browse Files
Download
Plain Diff
Merge branch 'master' into topic/basp
parents
56a38026
f11da3a1
Changes
10
Show whitespace changes
Inline
Side-by-side
Showing
10 changed files
with
994 additions
and
0 deletions
+994
-0
libcaf_net/CMakeLists.txt
libcaf_net/CMakeLists.txt
+3
-0
libcaf_net/caf/detail/convert_ip_endpoint.hpp
libcaf_net/caf/detail/convert_ip_endpoint.hpp
+32
-0
libcaf_net/caf/net/fwd.hpp
libcaf_net/caf/net/fwd.hpp
+2
-0
libcaf_net/caf/net/transport_worker_dispatcher.hpp
libcaf_net/caf/net/transport_worker_dispatcher.hpp
+153
-0
libcaf_net/caf/net/udp_datagram_socket.hpp
libcaf_net/caf/net/udp_datagram_socket.hpp
+83
-0
libcaf_net/src/convert_ip_endpoint.cpp
libcaf_net/src/convert_ip_endpoint.cpp
+63
-0
libcaf_net/src/udp_datagram_socket.cpp
libcaf_net/src/udp_datagram_socket.cpp
+140
-0
libcaf_net/test/convert_ip_endpoint.cpp
libcaf_net/test/convert_ip_endpoint.cpp
+112
-0
libcaf_net/test/transport_worker_dispatcher.cpp
libcaf_net/test/transport_worker_dispatcher.cpp
+293
-0
libcaf_net/test/udp_datagram_socket.cpp
libcaf_net/test/udp_datagram_socket.cpp
+113
-0
No files found.
libcaf_net/CMakeLists.txt
View file @
a6183018
...
...
@@ -10,6 +10,7 @@ set(LIBCAF_NET_SRCS
src/actor_proxy_impl.cpp
src/application.cpp
src/connection_state.cpp
src/convert_ip_endpoint.cpp
src/datagram_socket.cpp
src/ec.cpp
src/endpoint_manager.cpp
...
...
@@ -26,6 +27,8 @@ set(LIBCAF_NET_SRCS
src/stream_socket.cpp
src/tcp_accept_socket.cpp
src/tcp_stream_socket.cpp
src/tcp_stream_socket.cpp
src/udp_datagram_socket.cpp
)
add_custom_target
(
libcaf_net
)
...
...
libcaf_net/caf/detail/convert_ip_endpoint.hpp
0 → 100644
View file @
a6183018
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/socket_sys_includes.hpp"
#include "caf/fwd.hpp"
namespace
caf
{
namespace
detail
{
void
convert
(
const
ip_endpoint
&
src
,
sockaddr_storage
&
dst
);
error
convert
(
const
sockaddr_storage
&
src
,
ip_endpoint
&
dst
);
}
// namespace detail
}
// namespace caf
libcaf_net/caf/net/fwd.hpp
View file @
a6183018
...
...
@@ -29,6 +29,8 @@ class multiplexer;
class
socket_manager
;
template
<
class
Application
,
class
IdType
=
unit_t
>
class
transport_worker
;
template
<
class
Application
,
class
IdType
=
unit_t
>
class
transport_worker_dispatcher
;
struct
network_socket
;
struct
pipe_socket
;
...
...
libcaf_net/caf/net/transport_worker_dispatcher.hpp
0 → 100644
View file @
a6183018
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 <unordered_map>
#include "caf/byte.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/transport_worker.hpp"
#include "caf/net/write_packet_decorator.hpp"
#include "caf/span.hpp"
#include "caf/unit.hpp"
namespace
caf
{
namespace
net
{
/// implements a dispatcher that dispatches between transport and workers.
template
<
class
ApplicationFactory
,
class
IdType
>
class
transport_worker_dispatcher
{
public:
// -- member types -----------------------------------------------------------
using
id_type
=
IdType
;
using
factory_type
=
ApplicationFactory
;
using
application_type
=
typename
ApplicationFactory
::
application_type
;
using
worker_type
=
transport_worker
<
application_type
,
id_type
>
;
using
worker_ptr
=
transport_worker_ptr
<
application_type
,
id_type
>
;
// -- constructors, destructors, and assignment operators --------------------
transport_worker_dispatcher
(
factory_type
factory
)
:
factory_
(
std
::
move
(
factory
))
{
// nop
}
// -- member functions -------------------------------------------------------
template
<
class
Parent
>
error
init
(
Parent
&
)
{
CAF_ASSERT
(
workers_by_id_
.
empty
());
return
none
;
}
template
<
class
Parent
>
void
handle_data
(
Parent
&
parent
,
span
<
byte
>
data
,
id_type
id
)
{
auto
it
=
workers_by_id_
.
find
(
id
);
if
(
it
==
workers_by_id_
.
end
())
{
// TODO: where to get node_id from here?
add_new_worker
(
parent
,
node_id
{},
id
);
it
=
workers_by_id_
.
find
(
id
);
}
auto
worker
=
it
->
second
;
worker
->
handle_data
(
parent
,
data
);
}
template
<
class
Parent
>
void
write_message
(
Parent
&
parent
,
std
::
unique_ptr
<
net
::
endpoint_manager
::
message
>
msg
)
{
auto
sender
=
msg
->
msg
->
sender
;
if
(
!
sender
)
return
;
auto
nid
=
sender
->
node
();
auto
it
=
workers_by_node_
.
find
(
nid
);
if
(
it
==
workers_by_node_
.
end
())
{
// TODO: where to get id_type from here?
add_new_worker
(
parent
,
nid
,
id_type
{});
it
=
workers_by_node_
.
find
(
nid
);
}
auto
worker
=
it
->
second
;
worker
->
write_message
(
parent
,
std
::
move
(
msg
));
}
template
<
class
Parent
>
void
write_packet
(
Parent
&
parent
,
span
<
const
byte
>
header
,
span
<
const
byte
>
payload
,
id_type
id
)
{
parent
.
write_packet
(
header
,
payload
,
id
);
}
template
<
class
Parent
>
void
resolve
(
Parent
&
parent
,
const
std
::
string
&
path
,
actor
listener
)
{
// TODO path should be uri to lookup the corresponding worker
// if enpoint is known -> resolve actor through worker
// if not connect to endpoint?!
if
(
workers_by_id_
.
empty
())
return
;
auto
worker
=
workers_by_id_
.
begin
()
->
second
;
worker
->
resolve
(
parent
,
path
,
listener
);
}
template
<
class
...
Ts
>
void
set_timeout
(
uint64_t
timeout_id
,
id_type
id
,
Ts
&&
...)
{
workers_by_timeout_id_
.
emplace
(
timeout_id
,
workers_by_id_
.
at
(
id
));
}
template
<
class
Parent
>
void
timeout
(
Parent
&
parent
,
atom_value
value
,
uint64_t
id
)
{
auto
worker
=
workers_by_timeout_id_
.
at
(
id
);
worker
->
timeout
(
parent
,
value
,
id
);
workers_by_timeout_id_
.
erase
(
id
);
}
void
handle_error
(
sec
error
)
{
for
(
const
auto
&
p
:
workers_by_id_
)
{
auto
worker
=
p
.
second
;
worker
->
handle_error
(
error
);
}
}
template
<
class
Parent
>
error
add_new_worker
(
Parent
&
parent
,
node_id
node
,
id_type
id
)
{
auto
application
=
factory_
.
make
();
auto
worker
=
std
::
make_shared
<
worker_type
>
(
std
::
move
(
application
),
id
);
if
(
auto
err
=
worker
->
init
(
parent
))
return
err
;
workers_by_id_
.
emplace
(
std
::
move
(
id
),
worker
);
workers_by_node_
.
emplace
(
std
::
move
(
node
),
std
::
move
(
worker
));
return
none
;
}
private:
// -- worker lookups ---------------------------------------------------------
std
::
unordered_map
<
id_type
,
worker_ptr
>
workers_by_id_
;
std
::
unordered_map
<
node_id
,
worker_ptr
>
workers_by_node_
;
std
::
unordered_map
<
uint64_t
,
worker_ptr
>
workers_by_timeout_id_
;
factory_type
factory_
;
};
}
// namespace net
}
// namespace caf
libcaf_net/caf/net/udp_datagram_socket.hpp
0 → 100644
View file @
a6183018
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/fwd.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/net/network_socket.hpp"
namespace
caf
{
namespace
net
{
/// A datagram-oriented network communication endpoint for bidirectional
/// byte transmission.
struct
udp_datagram_socket
:
abstract_socket
<
udp_datagram_socket
>
{
using
super
=
abstract_socket
<
udp_datagram_socket
>
;
using
super
::
super
;
constexpr
operator
socket
()
const
noexcept
{
return
socket
{
id
};
}
constexpr
operator
network_socket
()
const
noexcept
{
return
network_socket
{
id
};
}
};
/// Creates a `udp_datagram_socket` bound to given port.
/// @param node ip_endpoint that contains the port to bind to. Pass port '0' to
/// bind to any unused port - The endpoint will be updated with the specific
/// port that was bound.
/// @returns The connected socket or an error.
/// @relates udp_datagram_socket
expected
<
std
::
pair
<
udp_datagram_socket
,
uint16_t
>>
make_udp_datagram_socket
(
ip_endpoint
ep
,
bool
reuse_addr
=
false
);
/// Enables or disables `SIO_UDP_CONNRESET` error on `x`.
/// @relates udp_datagram_socket
error
allow_connreset
(
udp_datagram_socket
x
,
bool
new_value
);
/// Receives the next datagram on socket `x`.
/// @param x The UDP socket for receiving datagrams.
/// @param buf Writable output buffer.
/// @returns The number of received bytes and the sender as `ip_endpoint` on
/// success, an error code otherwise.
/// @relates udp_datagram_socket
/// @post buf was modified and the resulting integer represents the length of
/// the received datagram, even if it did not fit into the given buffer.
variant
<
std
::
pair
<
size_t
,
ip_endpoint
>
,
sec
>
read
(
udp_datagram_socket
x
,
span
<
byte
>
buf
);
/// Sends the content of `buf` as a datagram to the endpoint `ep` on socket `x`.
/// @param x The UDP socket for sending datagrams.
/// @param buf The buffer to send.
/// @returns The number of written bytes on success, otherwise an error code.
/// @relates udp_datagram_socket
variant
<
size_t
,
sec
>
write
(
udp_datagram_socket
x
,
span
<
const
byte
>
buf
,
ip_endpoint
ep
);
/// Converts the result from I/O operation on a ::udp_datagram_socket to either
/// an error code or a non-zero positive integer.
/// @relates udp_datagram_socket
variant
<
size_t
,
sec
>
check_udp_datagram_socket_io_res
(
std
::
make_signed
<
size_t
>::
type
res
);
}
// namespace net
}
// namespace caf
libcaf_net/src/convert_ip_endpoint.cpp
0 → 100644
View file @
a6183018
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/detail/convert_ip_endpoint.hpp"
#include "caf/error.hpp"
#include "caf/ipv4_endpoint.hpp"
#include "caf/ipv6_endpoint.hpp"
#include "caf/sec.hpp"
namespace
caf
{
namespace
detail
{
void
convert
(
const
ip_endpoint
&
src
,
sockaddr_storage
&
dst
)
{
if
(
src
.
address
().
embeds_v4
())
{
auto
sockaddr4
=
reinterpret_cast
<
sockaddr_in
*>
(
&
dst
);
sockaddr4
->
sin_family
=
AF_INET
;
sockaddr4
->
sin_port
=
ntohs
(
src
.
port
());
sockaddr4
->
sin_addr
.
s_addr
=
src
.
address
().
embedded_v4
().
bits
();
}
else
{
auto
sockaddr6
=
reinterpret_cast
<
sockaddr_in6
*>
(
&
dst
);
sockaddr6
->
sin6_family
=
AF_INET6
;
sockaddr6
->
sin6_port
=
ntohs
(
src
.
port
());
memcpy
(
&
sockaddr6
->
sin6_addr
,
src
.
address
().
bytes
().
data
(),
src
.
address
().
bytes
().
size
());
}
}
error
convert
(
const
sockaddr_storage
&
src
,
ip_endpoint
&
dst
)
{
if
(
src
.
ss_family
==
AF_INET
)
{
auto
sockaddr4
=
reinterpret_cast
<
const
sockaddr_in
&>
(
src
);
ipv4_address
ipv4_addr
;
memcpy
(
ipv4_addr
.
data
().
data
(),
&
sockaddr4
.
sin_addr
,
ipv4_addr
.
size
());
dst
=
ip_endpoint
{
ipv4_addr
,
htons
(
sockaddr4
.
sin_port
)};
}
else
if
(
src
.
ss_family
==
AF_INET6
)
{
auto
sockaddr6
=
reinterpret_cast
<
const
sockaddr_in6
&>
(
src
);
ipv6_address
ipv6_addr
;
memcpy
(
ipv6_addr
.
bytes
().
data
(),
&
sockaddr6
.
sin6_addr
,
ipv6_addr
.
bytes
().
size
());
dst
=
ip_endpoint
{
ipv6_addr
,
htons
(
sockaddr6
.
sin6_port
)};
}
else
{
return
sec
::
invalid_argument
;
}
return
none
;
}
}
// namespace detail
}
// namespace caf
libcaf_net/src/udp_datagram_socket.cpp
0 → 100644
View file @
a6183018
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/udp_datagram_socket.hpp"
#include "caf/byte.hpp"
#include "caf/detail/convert_ip_endpoint.hpp"
#include "caf/detail/net_syscall.hpp"
#include "caf/detail/socket_sys_aliases.hpp"
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/expected.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/logger.hpp"
#include "caf/net/socket_guard.hpp"
#include "caf/span.hpp"
namespace
caf
{
namespace
net
{
#ifdef CAF_WINDOWS
error
allow_connreset
(
udp_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
(
udp_datagram_socket
x
,
bool
)
{
if
(
socket_cast
<
net
::
socket
>
(
x
)
==
invalid_socket
)
return
sec
::
socket_invalid
;
// nop; SIO_UDP_CONNRESET only exists on Windows
return
none
;
}
#endif // CAF_WINDOWS
expected
<
std
::
pair
<
udp_datagram_socket
,
uint16_t
>>
make_udp_datagram_socket
(
ip_endpoint
ep
,
bool
reuse_addr
)
{
CAF_LOG_TRACE
(
CAF_ARG
(
ep
));
sockaddr_storage
addr
;
detail
::
convert
(
ep
,
addr
);
CAF_NET_SYSCALL
(
"socket"
,
fd
,
==
,
invalid_socket_id
,
::
socket
(
addr
.
ss_family
,
SOCK_DGRAM
,
0
));
udp_datagram_socket
sock
{
fd
};
auto
sguard
=
make_socket_guard
(
sock
);
socklen_t
len
=
(
addr
.
ss_family
==
AF_INET
)
?
sizeof
(
sockaddr_in
)
:
sizeof
(
sockaddr_in6
);
if
(
reuse_addr
)
{
int
on
=
1
;
CAF_NET_SYSCALL
(
"setsockopt"
,
tmp1
,
!=
,
0
,
setsockopt
(
fd
,
SOL_SOCKET
,
SO_REUSEADDR
,
reinterpret_cast
<
setsockopt_ptr
>
(
&
on
),
static_cast
<
socket_size_type
>
(
sizeof
(
on
))));
}
CAF_NET_SYSCALL
(
"bind"
,
err
,
!=
,
0
,
::
bind
(
sock
.
id
,
reinterpret_cast
<
sockaddr
*>
(
&
addr
),
len
));
CAF_NET_SYSCALL
(
"getsockname"
,
erro
,
!=
,
0
,
getsockname
(
sock
.
id
,
reinterpret_cast
<
sockaddr
*>
(
&
addr
),
&
len
));
CAF_LOG_DEBUG
(
CAF_ARG
(
sock
.
id
));
auto
port
=
addr
.
ss_family
==
AF_INET
?
reinterpret_cast
<
sockaddr_in
*>
(
&
addr
)
->
sin_port
:
reinterpret_cast
<
sockaddr_in6
*>
(
&
addr
)
->
sin6_port
;
return
std
::
make_pair
(
sguard
.
release
(),
port
);
}
variant
<
std
::
pair
<
size_t
,
ip_endpoint
>
,
sec
>
read
(
udp_datagram_socket
x
,
span
<
byte
>
buf
)
{
sockaddr_storage
addr
=
{};
socklen_t
len
=
sizeof
(
sockaddr_storage
);
auto
res
=
::
recvfrom
(
x
.
id
,
reinterpret_cast
<
socket_recv_ptr
>
(
buf
.
data
()),
buf
.
size
(),
0
,
reinterpret_cast
<
sockaddr
*>
(
&
addr
),
&
len
);
auto
ret
=
check_udp_datagram_socket_io_res
(
res
);
if
(
auto
num_bytes
=
get_if
<
size_t
>
(
&
ret
))
{
CAF_LOG_INFO_IF
(
*
num_bytes
==
0
,
"Received empty datagram"
);
CAF_LOG_WARNING_IF
(
*
num_bytes
>
buf
.
size
(),
"recvfrom cut of message, only received "
<<
CAF_ARG
(
buf
.
size
())
<<
" of "
<<
CAF_ARG
(
num_bytes
)
<<
" bytes"
);
ip_endpoint
ep
;
// TODO: how to properly pass error further?
if
(
auto
err
=
detail
::
convert
(
addr
,
ep
))
return
sec
::
runtime_error
;
return
std
::
pair
<
size_t
,
ip_endpoint
>
(
*
num_bytes
,
ep
);
}
else
{
return
get
<
sec
>
(
ret
);
}
}
variant
<
size_t
,
sec
>
write
(
udp_datagram_socket
x
,
span
<
const
byte
>
buf
,
ip_endpoint
ep
)
{
sockaddr_storage
addr
;
detail
::
convert
(
ep
,
addr
);
auto
res
=
::
sendto
(
x
.
id
,
reinterpret_cast
<
socket_send_ptr
>
(
buf
.
data
()),
buf
.
size
(),
0
,
reinterpret_cast
<
sockaddr
*>
(
&
addr
),
ep
.
address
().
embeds_v4
()
?
sizeof
(
sockaddr_in
)
:
sizeof
(
sockaddr_in6
));
auto
ret
=
check_udp_datagram_socket_io_res
(
res
);
if
(
auto
num_bytes
=
get_if
<
size_t
>
(
&
ret
))
return
*
num_bytes
;
else
return
get
<
sec
>
(
ret
);
}
variant
<
size_t
,
sec
>
check_udp_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/test/convert_ip_endpoint.cpp
0 → 100644
View file @
a6183018
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 convert_ip_endpoint
#include "caf/detail/convert_ip_endpoint.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include <cstring>
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/ipv4_endpoint.hpp"
#include "caf/ipv6_endpoint.hpp"
using
namespace
caf
;
using
namespace
caf
::
detail
;
namespace
{
struct
fixture
:
host_fixture
{
fixture
()
:
host_fixture
()
{
memset
(
&
sockaddr6_src
,
0
,
sizeof
(
sockaddr_in6
));
memset
(
&
sockaddr6_dst
,
0
,
sizeof
(
sockaddr_in6
));
sockaddr6_src
.
sin6_family
=
AF_INET6
;
sockaddr6_src
.
sin6_port
=
htons
(
23
);
sockaddr6_src
.
sin6_addr
=
in6addr_loopback
;
memset
(
&
sockaddr4_src
,
0
,
sizeof
(
sockaddr_in
));
memset
(
&
sockaddr4_dst
,
0
,
sizeof
(
sockaddr_in
));
sockaddr4_src
.
sin_family
=
AF_INET
;
sockaddr4_src
.
sin_port
=
htons
(
23
);
sockaddr4_src
.
sin_addr
.
s_addr
=
INADDR_LOOPBACK
;
}
sockaddr_in6
sockaddr6_src
;
sockaddr_in6
sockaddr6_dst
;
sockaddr_in
sockaddr4_src
;
sockaddr_in
sockaddr4_dst
;
ip_endpoint
ep_src
;
ip_endpoint
ep_dst
;
};
}
// namespace
CAF_TEST_FIXTURE_SCOPE
(
convert_ip_endpoint_tests
,
fixture
)
CAF_TEST
(
sockaddr_in6
roundtrip
)
{
ip_endpoint
ep
;
CAF_MESSAGE
(
"converting sockaddr_in6 to ip_endpoint"
);
CAF_CHECK_EQUAL
(
convert
(
reinterpret_cast
<
sockaddr_storage
&>
(
sockaddr6_src
),
ep
),
none
);
CAF_MESSAGE
(
"converting ip_endpoint to sockaddr_in6"
);
convert
(
ep
,
reinterpret_cast
<
sockaddr_storage
&>
(
sockaddr6_dst
));
CAF_CHECK_EQUAL
(
memcmp
(
&
sockaddr6_src
,
&
sockaddr6_dst
,
sizeof
(
sockaddr_in6
)),
0
);
}
CAF_TEST
(
ipv6_endpoint
roundtrip
)
{
sockaddr_storage
addr
;
memset
(
&
addr
,
0
,
sizeof
(
sockaddr_storage
));
if
(
auto
err
=
detail
::
parse
(
"[::1]:55555"
,
ep_src
))
CAF_FAIL
(
"unable to parse input: "
<<
err
);
CAF_MESSAGE
(
"converting ip_endpoint to sockaddr_in6"
);
convert
(
ep_src
,
addr
);
CAF_MESSAGE
(
"converting sockaddr_in6 to ip_endpoint"
);
CAF_CHECK_EQUAL
(
convert
(
addr
,
ep_dst
),
none
);
CAF_CHECK_EQUAL
(
ep_src
,
ep_dst
);
}
CAF_TEST
(
sockaddr_in4
roundtrip
)
{
ip_endpoint
ep
;
CAF_MESSAGE
(
"converting sockaddr_in to ip_endpoint"
);
CAF_CHECK_EQUAL
(
convert
(
reinterpret_cast
<
sockaddr_storage
&>
(
sockaddr4_src
),
ep
),
none
);
CAF_MESSAGE
(
"converting ip_endpoint to sockaddr_in"
);
convert
(
ep
,
reinterpret_cast
<
sockaddr_storage
&>
(
sockaddr4_dst
));
CAF_CHECK_EQUAL
(
memcmp
(
&
sockaddr4_src
,
&
sockaddr4_dst
,
sizeof
(
sockaddr_in
)),
0
);
}
CAF_TEST
(
ipv4_endpoint
roundtrip
)
{
sockaddr_storage
addr
;
memset
(
&
addr
,
0
,
sizeof
(
sockaddr_storage
));
if
(
auto
err
=
detail
::
parse
(
"127.0.0.1:55555"
,
ep_src
))
CAF_FAIL
(
"unable to parse input: "
<<
err
);
CAF_MESSAGE
(
"converting ip_endpoint to sockaddr_in"
);
convert
(
ep_src
,
addr
);
CAF_MESSAGE
(
"converting sockaddr_in to ip_endpoint"
);
CAF_CHECK_EQUAL
(
convert
(
addr
,
ep_dst
),
none
);
CAF_CHECK_EQUAL
(
ep_src
,
ep_dst
);
}
CAF_TEST_FIXTURE_SCOPE_END
();
libcaf_net/test/transport_worker_dispatcher.cpp
0 → 100644
View file @
a6183018
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 transport_worker_dispatcher
#include "caf/net/transport_worker_dispatcher.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include "caf/make_actor.hpp"
#include "caf/monitorable_actor.hpp"
#include "caf/node_id.hpp"
#include "caf/uri.hpp"
using
namespace
caf
;
using
namespace
caf
::
net
;
namespace
{
constexpr
string_view
hello_test
=
"hello_test"
;
struct
dummy_actor
:
public
monitorable_actor
{
dummy_actor
(
actor_config
&
cfg
)
:
monitorable_actor
(
cfg
)
{
// nop
}
void
enqueue
(
mailbox_element_ptr
,
execution_unit
*
)
override
{
// nop
}
};
class
dummy_application
{
public:
dummy_application
(
std
::
shared_ptr
<
std
::
vector
<
byte
>>
rec_buf
,
uint8_t
id
)
:
rec_buf_
(
std
::
move
(
rec_buf
)),
id_
(
id
){
// nop
};
~
dummy_application
()
=
default
;
template
<
class
Parent
>
error
init
(
Parent
&
)
{
rec_buf_
->
push_back
(
static_cast
<
byte
>
(
id_
));
return
none
;
}
template
<
class
Transport
>
void
write_message
(
Transport
&
transport
,
std
::
unique_ptr
<
endpoint_manager
::
message
>
msg
)
{
rec_buf_
->
push_back
(
static_cast
<
byte
>
(
id_
));
transport
.
write_packet
(
span
<
byte
>
{},
make_span
(
msg
->
payload
));
}
template
<
class
Parent
>
void
handle_data
(
Parent
&
,
span
<
const
byte
>
)
{
rec_buf_
->
push_back
(
static_cast
<
byte
>
(
id_
));
}
template
<
class
Manager
>
void
resolve
(
Manager
&
,
const
std
::
string
&
,
actor
)
{
rec_buf_
->
push_back
(
static_cast
<
byte
>
(
id_
));
}
template
<
class
Transport
>
void
timeout
(
Transport
&
,
atom_value
,
uint64_t
)
{
rec_buf_
->
push_back
(
static_cast
<
byte
>
(
id_
));
}
void
handle_error
(
sec
)
{
rec_buf_
->
push_back
(
static_cast
<
byte
>
(
id_
));
}
static
expected
<
std
::
vector
<
byte
>>
serialize
(
actor_system
&
,
const
type_erased_tuple
&
)
{
return
std
::
vector
<
byte
>
{};
}
private:
std
::
shared_ptr
<
std
::
vector
<
byte
>>
rec_buf_
;
uint8_t
id_
;
};
struct
dummy_application_factory
{
public:
using
application_type
=
dummy_application
;
dummy_application_factory
(
std
::
shared_ptr
<
std
::
vector
<
byte
>>
buf
)
:
buf_
(
buf
),
application_cnt_
(
0
)
{
// nop
}
dummy_application
make
()
{
return
dummy_application
{
buf_
,
application_cnt_
++
};
}
private:
std
::
shared_ptr
<
std
::
vector
<
byte
>>
buf_
;
uint8_t
application_cnt_
;
};
struct
dummy_transport
{
using
transport_type
=
dummy_transport
;
using
application_type
=
dummy_application
;
dummy_transport
(
std
::
shared_ptr
<
std
::
vector
<
byte
>>
buf
)
:
buf_
(
buf
)
{
}
template
<
class
IdType
>
void
write_packet
(
span
<
const
byte
>
header
,
span
<
const
byte
>
payload
,
IdType
)
{
buf_
->
insert
(
buf_
->
end
(),
header
.
begin
(),
header
.
end
());
buf_
->
insert
(
buf_
->
end
(),
payload
.
begin
(),
payload
.
end
());
}
private:
std
::
shared_ptr
<
std
::
vector
<
byte
>>
buf_
;
};
struct
testdata
{
testdata
(
uint8_t
worker_id
,
node_id
id
,
ip_endpoint
ep
)
:
worker_id
(
worker_id
),
nid
(
id
),
ep
(
ep
)
{
// nop
}
uint8_t
worker_id
;
node_id
nid
;
ip_endpoint
ep
;
};
// TODO: switch to std::operator""s when switching to C++14
ip_endpoint
operator
""
_ep
(
const
char
*
cstr
,
size_t
cstr_len
)
{
ip_endpoint
ep
;
string_view
str
(
cstr
,
cstr_len
);
if
(
auto
err
=
parse
(
str
,
ep
))
CAF_FAIL
(
"parse returned error: "
<<
err
);
return
ep
;
}
uri
operator
""
_u
(
const
char
*
cstr
,
size_t
cstr_len
)
{
uri
result
;
string_view
str
{
cstr
,
cstr_len
};
auto
err
=
parse
(
str
,
result
);
if
(
err
)
CAF_FAIL
(
"error while parsing "
<<
str
<<
": "
<<
to_string
(
err
));
return
result
;
}
struct
fixture
:
host_fixture
{
using
dispatcher_type
=
transport_worker_dispatcher
<
dummy_application_factory
,
ip_endpoint
>
;
fixture
()
:
buf
{
std
::
make_shared
<
std
::
vector
<
byte
>>
()},
dispatcher
{
dummy_application_factory
{
buf
}},
dummy
{
buf
}
{
add_new_workers
();
}
std
::
unique_ptr
<
net
::
endpoint_manager
::
message
>
make_dummy_message
(
node_id
nid
)
{
actor_id
aid
=
42
;
actor_config
cfg
;
auto
p
=
make_actor
<
dummy_actor
,
strong_actor_ptr
>
(
aid
,
nid
,
&
sys
,
cfg
);
auto
test_span
=
as_bytes
(
make_span
(
hello_test
));
std
::
vector
<
byte
>
payload
(
test_span
.
begin
(),
test_span
.
end
());
auto
strong_actor
=
actor_cast
<
strong_actor_ptr
>
(
p
);
mailbox_element
::
forwarding_stack
stack
;
auto
elem
=
make_mailbox_element
(
std
::
move
(
strong_actor
),
make_message_id
(
12345
),
std
::
move
(
stack
),
make_message
());
return
detail
::
make_unique
<
endpoint_manager
::
message
>
(
std
::
move
(
elem
),
payload
);
}
bool
contains
(
byte
x
)
{
return
std
::
count
(
buf
->
begin
(),
buf
->
end
(),
x
)
>
0
;
}
void
add_new_workers
()
{
for
(
auto
&
data
:
test_data
)
{
if
(
auto
err
=
dispatcher
.
add_new_worker
(
dummy
,
data
.
nid
,
data
.
ep
))
CAF_FAIL
(
"add_new_worker returned an error: "
<<
err
);
}
buf
->
clear
();
}
void
test_write_message
(
testdata
&
testcase
)
{
auto
msg
=
make_dummy_message
(
testcase
.
nid
);
if
(
!
msg
->
msg
->
sender
)
CAF_FAIL
(
"sender is null"
);
dispatcher
.
write_message
(
dummy
,
std
::
move
(
msg
));
}
actor_system_config
cfg
{};
actor_system
sys
{
cfg
};
std
::
shared_ptr
<
std
::
vector
<
byte
>>
buf
;
dispatcher_type
dispatcher
;
dummy_transport
dummy
;
std
::
vector
<
testdata
>
test_data
{
{
0
,
make_node_id
(
"http:file"
_u
),
"[::1]:1"
_ep
},
{
1
,
make_node_id
(
"http:file?a=1&b=2"
_u
),
"[fe80::2:34]:12345"
_ep
},
{
2
,
make_node_id
(
"http:file#42"
_u
),
"[1234::17]:4444"
_ep
},
{
3
,
make_node_id
(
"http:file?a=1&b=2#42"
_u
),
"[2332::1]:12"
_ep
},
};
};
#define CHECK_HANDLE_DATA(testcase) \
dispatcher.handle_data(dummy, span<byte>{}, testcase.ep); \
CAF_CHECK_EQUAL(buf->size(), 1u); \
CAF_CHECK_EQUAL(static_cast<byte>(testcase.worker_id), buf->at(0)); \
buf->clear();
#define CHECK_WRITE_MESSAGE(testcase) \
test_write_message(testcase); \
CAF_CHECK_EQUAL(buf->size(), hello_test.size() + 1u); \
CAF_CHECK_EQUAL(static_cast<byte>(testcase.worker_id), buf->at(0)); \
CAF_CHECK_EQUAL(memcmp(buf->data() + 1, hello_test.data(), \
hello_test.size()), \
0); \
buf->clear();
#define CHECK_TIMEOUT(testcase) \
dispatcher.set_timeout(1u, testcase.ep); \
dispatcher.timeout(dummy, atom("dummy"), 1u); \
CAF_CHECK_EQUAL(buf->size(), 1u); \
CAF_CHECK_EQUAL(static_cast<byte>(testcase.worker_id), buf->at(0)); \
buf->clear();
}
// namespace
CAF_TEST_FIXTURE_SCOPE
(
transport_worker_dispatcher_test
,
fixture
)
CAF_TEST
(
init
)
{
dispatcher_type
dispatcher
{
dummy_application_factory
{
buf
}};
if
(
auto
err
=
dispatcher
.
init
(
dummy
))
CAF_FAIL
(
"init failed with error: "
<<
err
);
}
CAF_TEST
(
handle_data
)
{
CHECK_HANDLE_DATA
(
test_data
.
at
(
0
));
CHECK_HANDLE_DATA
(
test_data
.
at
(
1
));
CHECK_HANDLE_DATA
(
test_data
.
at
(
2
));
CHECK_HANDLE_DATA
(
test_data
.
at
(
3
));
}
CAF_TEST
(
write_message
write_packet
)
{
CHECK_WRITE_MESSAGE
(
test_data
.
at
(
0
));
CHECK_WRITE_MESSAGE
(
test_data
.
at
(
1
));
CHECK_WRITE_MESSAGE
(
test_data
.
at
(
2
));
CHECK_WRITE_MESSAGE
(
test_data
.
at
(
3
));
}
CAF_TEST
(
resolve
)
{
// TODO think of a test for this
}
CAF_TEST
(
timeout
)
{
CHECK_TIMEOUT
(
test_data
.
at
(
0
));
CHECK_TIMEOUT
(
test_data
.
at
(
1
));
CHECK_TIMEOUT
(
test_data
.
at
(
2
));
CHECK_TIMEOUT
(
test_data
.
at
(
3
));
}
CAF_TEST
(
handle_error
)
{
dispatcher
.
handle_error
(
sec
::
unavailable_or_would_block
);
CAF_CHECK_EQUAL
(
buf
->
size
(),
4u
);
CAF_CHECK
(
contains
(
byte
(
0
)));
CAF_CHECK
(
contains
(
byte
(
1
)));
CAF_CHECK
(
contains
(
byte
(
2
)));
CAF_CHECK
(
contains
(
byte
(
3
)));
}
CAF_TEST_FIXTURE_SCOPE_END
()
libcaf_net/test/udp_datagram_socket.cpp
0 → 100644
View file @
a6183018
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 udp_datagram_socket
#include "caf/net/udp_datagram_socket.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include "caf/detail/net_syscall.hpp"
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/ip_address.hpp"
#include "caf/ipv4_address.hpp"
#include "caf/net/ip.hpp"
#include "caf/net/socket_guard.hpp"
using
namespace
caf
;
using
namespace
caf
::
net
;
namespace
{
constexpr
string_view
hello_test
=
"Hello test!"
;
struct
fixture
:
host_fixture
{
fixture
()
:
host_fixture
(),
v4_local
{
make_ipv4_address
(
127
,
0
,
0
,
1
)},
v6_local
{{
0
},
{
0x1
}},
// TODO: use local_addresses() when merged
addrs
{
net
::
ip
::
resolve
(
"localhost"
)}
{
}
bool
contains
(
ip_address
x
)
{
return
std
::
count
(
addrs
.
begin
(),
addrs
.
end
(),
x
)
>
0
;
}
void
test_send_receive
(
ip_address
addr
)
{
std
::
vector
<
byte
>
buf
(
1024
);
ip_endpoint
ep
(
addr
,
0
);
auto
send_pair
=
unbox
(
make_udp_datagram_socket
(
ep
));
auto
send_socket
=
send_pair
.
first
;
auto
receive_pair
=
unbox
(
make_udp_datagram_socket
(
ep
));
auto
receive_socket
=
receive_pair
.
first
;
ep
.
port
(
ntohs
(
receive_pair
.
second
));
CAF_MESSAGE
(
"sending data to: "
<<
to_string
(
ep
));
auto
send_guard
=
make_socket_guard
(
send_socket
);
auto
receive_guard
=
make_socket_guard
(
receive_socket
);
if
(
auto
err
=
nonblocking
(
socket_cast
<
net
::
socket
>
(
receive_socket
),
true
))
CAF_FAIL
(
"nonblocking returned an error"
<<
err
);
auto
test_read_res
=
read
(
receive_socket
,
make_span
(
buf
));
if
(
auto
err
=
get_if
<
sec
>
(
&
test_read_res
))
CAF_CHECK_EQUAL
(
*
err
,
sec
::
unavailable_or_would_block
);
else
CAF_FAIL
(
"read should have failed"
);
auto
write_ret
=
write
(
send_socket
,
as_bytes
(
make_span
(
hello_test
)),
ep
);
if
(
auto
num_bytes
=
get_if
<
size_t
>
(
&
write_ret
))
CAF_CHECK_EQUAL
(
*
num_bytes
,
hello_test
.
size
());
else
CAF_FAIL
(
"write returned an error: "
<<
sys
.
render
(
get
<
sec
>
(
write_ret
)));
auto
read_ret
=
read
(
receive_socket
,
make_span
(
buf
));
if
(
auto
read_res
=
get_if
<
std
::
pair
<
size_t
,
ip_endpoint
>>
(
&
read_ret
))
{
CAF_CHECK_EQUAL
(
read_res
->
first
,
hello_test
.
size
());
buf
.
resize
(
read_res
->
first
);
}
else
{
CAF_FAIL
(
"write returned an error: "
<<
sys
.
render
(
get
<
sec
>
(
read_ret
)));
}
string_view
buf_view
{
reinterpret_cast
<
const
char
*>
(
buf
.
data
()),
buf
.
size
()};
CAF_CHECK_EQUAL
(
buf_view
,
hello_test
);
}
actor_system_config
cfg
;
actor_system
sys
{
cfg
};
ip_address
v4_local
;
ip_address
v6_local
;
std
::
vector
<
ip_address
>
addrs
;
};
}
// namespace
CAF_TEST_FIXTURE_SCOPE
(
udp_datagram_socket_test
,
fixture
)
CAF_TEST
(
send
and
receive
)
{
// TODO: check which versions exist and test existing versions accordingly
// -> local_addresses()
if
(
contains
(
v4_local
))
{
test_send_receive
(
v4_local
);
}
else
if
(
contains
(
v6_local
))
{
test_send_receive
(
v6_local
);
}
else
{
CAF_FAIL
(
"could not resolve 'localhost'"
);
}
}
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