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
c42afd22
Commit
c42afd22
authored
Oct 31, 2019
by
Dominik Charousset
Committed by
GitHub
Oct 31, 2019
Browse files
Options
Browse Files
Download
Plain Diff
Merge pull request #35
Streamline stream_transport test
parents
54c92bbd
faa9e0f0
Changes
3
Hide whitespace changes
Inline
Side-by-side
Showing
3 changed files
with
94 additions
and
43 deletions
+94
-43
libcaf_net/caf/net/socket_guard.hpp
libcaf_net/caf/net/socket_guard.hpp
+29
-2
libcaf_net/test/socket_guard.cpp
libcaf_net/test/socket_guard.cpp
+19
-7
libcaf_net/test/stream_transport.cpp
libcaf_net/test/stream_transport.cpp
+46
-34
No files found.
libcaf_net/caf/net/socket_guard.hpp
View file @
c42afd22
...
@@ -27,21 +27,48 @@ namespace net {
...
@@ -27,21 +27,48 @@ namespace net {
template
<
class
Socket
>
template
<
class
Socket
>
class
socket_guard
{
class
socket_guard
{
public:
public:
explicit
socket_guard
(
Socket
sock
)
:
sock_
(
sock
)
{
socket_guard
()
noexcept
:
sock_
(
invalid_socket_id
)
{
// nop
// nop
}
}
explicit
socket_guard
(
Socket
sock
)
noexcept
:
sock_
(
sock
)
{
// nop
}
socket_guard
(
socket_guard
&&
other
)
noexcept
:
sock_
(
other
.
release
())
{
// nop
}
socket_guard
(
const
socket_guard
&
)
=
delete
;
socket_guard
&
operator
=
(
socket_guard
&&
other
)
noexcept
{
reset
(
other
.
release
());
return
*
this
;
}
socket_guard
&
operator
=
(
const
socket_guard
&
)
=
delete
;
~
socket_guard
()
{
~
socket_guard
()
{
if
(
sock_
.
id
!=
invalid_socket_id
)
if
(
sock_
.
id
!=
invalid_socket_id
)
close
(
sock_
);
close
(
sock_
);
}
}
Socket
release
()
{
void
reset
(
Socket
x
)
noexcept
{
if
(
sock_
.
id
!=
invalid_socket_id
)
close
(
sock_
);
sock_
=
x
;
}
Socket
release
()
noexcept
{
auto
sock
=
sock_
;
auto
sock
=
sock_
;
sock_
.
id
=
invalid_socket_id
;
sock_
.
id
=
invalid_socket_id
;
return
sock
;
return
sock
;
}
}
Socket
socket
()
const
noexcept
{
return
sock_
;
}
private:
private:
Socket
sock_
;
Socket
sock_
;
};
};
...
...
libcaf_net/test/socket_guard.cpp
View file @
c42afd22
...
@@ -32,7 +32,7 @@ namespace {
...
@@ -32,7 +32,7 @@ namespace {
constexpr
socket_id
dummy_id
=
13
;
constexpr
socket_id
dummy_id
=
13
;
struct
dummy_socket
{
struct
dummy_socket
{
dummy_socket
(
socket_id
&
id
,
bool
&
closed
)
:
id
(
id
),
closed
(
closed
)
{
dummy_socket
(
socket_id
id
,
bool
&
closed
)
:
id
(
id
),
closed
(
closed
)
{
// nop
// nop
}
}
...
@@ -44,20 +44,20 @@ struct dummy_socket {
...
@@ -44,20 +44,20 @@ struct dummy_socket {
return
*
this
;
return
*
this
;
}
}
socket_id
&
id
;
socket_id
id
;
bool
&
closed
;
bool
&
closed
;
};
};
void
close
(
dummy_socket
x
)
{
void
close
(
dummy_socket
x
)
{
x
.
id
=
invalid_socket_id
;
x
.
closed
=
true
;
x
.
closed
=
true
;
}
}
struct
fixture
{
struct
fixture
{
fixture
()
:
id
{
dummy_id
},
closed
{
false
},
sock
{
id
,
closed
}
{
fixture
()
:
closed
{
false
},
sock
{
dummy_
id
,
closed
}
{
// nop
// nop
}
}
socket_id
id
;
bool
closed
;
bool
closed
;
dummy_socket
sock
;
dummy_socket
sock
;
};
};
...
@@ -69,17 +69,29 @@ CAF_TEST_FIXTURE_SCOPE(socket_guard_tests, fixture)
...
@@ -69,17 +69,29 @@ CAF_TEST_FIXTURE_SCOPE(socket_guard_tests, fixture)
CAF_TEST
(
cleanup
)
{
CAF_TEST
(
cleanup
)
{
{
{
auto
guard
=
make_socket_guard
(
sock
);
auto
guard
=
make_socket_guard
(
sock
);
CAF_CHECK_EQUAL
(
sock
.
id
,
dummy_id
);
CAF_CHECK_EQUAL
(
guard
.
socket
()
.
id
,
dummy_id
);
}
}
CAF_CHECK
(
sock
.
closed
);
CAF_CHECK
(
sock
.
closed
);
}
}
CAF_TEST
(
reset
)
{
{
auto
guard
=
make_socket_guard
(
sock
);
CAF_CHECK_EQUAL
(
guard
.
socket
().
id
,
dummy_id
);
guard
.
release
();
CAF_CHECK_EQUAL
(
guard
.
socket
().
id
,
invalid_socket_id
);
guard
.
reset
(
sock
);
CAF_CHECK_EQUAL
(
guard
.
socket
().
id
,
dummy_id
);
}
CAF_CHECK_EQUAL
(
sock
.
closed
,
true
);
}
CAF_TEST
(
release
)
{
CAF_TEST
(
release
)
{
{
{
auto
guard
=
make_socket_guard
(
sock
);
auto
guard
=
make_socket_guard
(
sock
);
CAF_CHECK_EQUAL
(
sock
.
id
,
dummy_id
);
CAF_CHECK_EQUAL
(
guard
.
socket
()
.
id
,
dummy_id
);
guard
.
release
();
guard
.
release
();
CAF_CHECK_EQUAL
(
sock
.
id
,
invalid_socket_id
);
CAF_CHECK_EQUAL
(
guard
.
socket
()
.
id
,
invalid_socket_id
);
}
}
CAF_CHECK_EQUAL
(
sock
.
closed
,
false
);
CAF_CHECK_EQUAL
(
sock
.
closed
,
false
);
}
}
...
...
libcaf_net/test/stream_transport.cpp
View file @
c42afd22
...
@@ -29,8 +29,10 @@
...
@@ -29,8 +29,10 @@
#include "caf/make_actor.hpp"
#include "caf/make_actor.hpp"
#include "caf/net/actor_proxy_impl.hpp"
#include "caf/net/actor_proxy_impl.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/endpoint_manager_impl.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/socket_guard.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/serializer_impl.hpp"
#include "caf/serializer_impl.hpp"
#include "caf/span.hpp"
#include "caf/span.hpp"
...
@@ -39,15 +41,24 @@ using namespace caf;
...
@@ -39,15 +41,24 @@ using namespace caf;
using
namespace
caf
::
net
;
using
namespace
caf
::
net
;
namespace
{
namespace
{
constexpr
string_view
hello_manager
=
"hello manager!"
;
constexpr
string_view
hello_manager
=
"hello manager!"
;
struct
fixture
:
test_coordinator_fixture
<>
,
host_fixture
{
struct
fixture
:
test_coordinator_fixture
<>
,
host_fixture
{
fixture
()
{
using
buffer_type
=
std
::
vector
<
byte
>
;
using
buffer_ptr
=
std
::
shared_ptr
<
buffer_type
>
;
fixture
()
:
recv_buf
(
1024
),
shared_buf
{
std
::
make_shared
<
buffer_type
>
()}
{
mpx
=
std
::
make_shared
<
multiplexer
>
();
mpx
=
std
::
make_shared
<
multiplexer
>
();
if
(
auto
err
=
mpx
->
init
())
if
(
auto
err
=
mpx
->
init
())
CAF_FAIL
(
"mpx->init failed: "
<<
sys
.
render
(
err
));
CAF_FAIL
(
"mpx->init failed: "
<<
sys
.
render
(
err
));
mpx
->
set_thread_id
();
mpx
->
set_thread_id
();
CAF_CHECK_EQUAL
(
mpx
->
num_socket_managers
(),
1u
);
auto
sockets
=
unbox
(
make_stream_socket_pair
());
send_socket_guard
.
reset
(
sockets
.
first
);
recv_socket_guard
.
reset
(
sockets
.
second
);
if
(
auto
err
=
nonblocking
(
recv_socket_guard
.
socket
(),
true
))
CAF_FAIL
(
"nonblocking returned an error: "
<<
err
);
}
}
bool
handle_io_event
()
override
{
bool
handle_io_event
()
override
{
...
@@ -55,11 +66,19 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
...
@@ -55,11 +66,19 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
}
}
multiplexer_ptr
mpx
;
multiplexer_ptr
mpx
;
buffer_type
recv_buf
;
socket_guard
<
stream_socket
>
send_socket_guard
;
socket_guard
<
stream_socket
>
recv_socket_guard
;
buffer_ptr
shared_buf
;
};
};
class
dummy_application
{
class
dummy_application
{
using
buffer_type
=
std
::
vector
<
byte
>
;
using
buffer_ptr
=
std
::
shared_ptr
<
buffer_type
>
;
public:
public:
dummy_application
(
std
::
shared_ptr
<
std
::
vector
<
byte
>>
rec_buf
)
dummy_application
(
buffer_ptr
rec_buf
)
:
rec_buf_
(
std
::
move
(
rec_buf
)){
:
rec_buf_
(
std
::
move
(
rec_buf
)){
// nop
// nop
};
};
...
@@ -118,17 +137,17 @@ public:
...
@@ -118,17 +137,17 @@ public:
// nop
// nop
}
}
static
expected
<
std
::
vector
<
byte
>
>
serialize
(
actor_system
&
sys
,
static
expected
<
buffer_type
>
serialize
(
actor_system
&
sys
,
const
type_erased_tuple
&
x
)
{
const
type_erased_tuple
&
x
)
{
std
::
vector
<
byte
>
result
;
buffer_type
result
;
serializer_impl
<
std
::
vector
<
byte
>
>
sink
{
sys
,
result
};
serializer_impl
<
buffer_type
>
sink
{
sys
,
result
};
if
(
auto
err
=
message
::
save
(
sink
,
x
))
if
(
auto
err
=
message
::
save
(
sink
,
x
))
return
err
;
return
err
;
return
result
;
return
result
;
}
}
private:
private:
std
::
shared_ptr
<
std
::
vector
<
byte
>>
rec_buf_
;
buffer_ptr
rec_buf_
;
};
};
}
// namespace
}
// namespace
...
@@ -137,39 +156,32 @@ CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture)
...
@@ -137,39 +156,32 @@ CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture)
CAF_TEST
(
receive
)
{
CAF_TEST
(
receive
)
{
using
transport_type
=
stream_transport
<
dummy_application
>
;
using
transport_type
=
stream_transport
<
dummy_application
>
;
std
::
vector
<
byte
>
read_buf
(
1024
);
auto
mgr
=
make_endpoint_manager
(
mpx
,
sys
,
CAF_CHECK_EQUAL
(
mpx
->
num_socket_managers
(),
1u
);
transport_type
{
recv_socket_guard
.
release
(),
auto
buf
=
std
::
make_shared
<
std
::
vector
<
byte
>>
();
dummy_application
{
auto
sockets
=
unbox
(
make_stream_socket_pair
());
shared_buf
}});
if
(
auto
err
=
nonblocking
(
sockets
.
second
,
true
))
CAF_FAIL
(
"nonblocking() returned an error: "
<<
err
);
CAF_CHECK_EQUAL
(
read
(
sockets
.
second
,
make_span
(
read_buf
)),
sec
::
unavailable_or_would_block
);
auto
guard
=
detail
::
make_scope_guard
([
&
]
{
close
(
sockets
.
second
);
});
transport_type
transport
{
sockets
.
first
,
dummy_application
{
buf
}};
transport
.
configure_read
(
net
::
receive_policy
::
exactly
(
hello_manager
.
size
()));
auto
mgr
=
make_endpoint_manager
(
mpx
,
sys
,
transport
);
CAF_CHECK_EQUAL
(
mgr
->
init
(),
none
);
CAF_CHECK_EQUAL
(
mgr
->
init
(),
none
);
auto
mgr_impl
=
mgr
.
downcast
<
endpoint_manager_impl
<
transport_type
>>
();
CAF_CHECK
(
mgr_impl
!=
nullptr
);
auto
&
transport
=
mgr_impl
->
transport
();
transport
.
configure_read
(
receive_policy
::
exactly
(
hello_manager
.
size
()));
CAF_CHECK_EQUAL
(
mpx
->
num_socket_managers
(),
2u
);
CAF_CHECK_EQUAL
(
mpx
->
num_socket_managers
(),
2u
);
CAF_CHECK_EQUAL
(
write
(
sockets
.
second
,
as_bytes
(
make_span
(
hello_manager
))),
CAF_CHECK_EQUAL
(
write
(
send_socket_guard
.
socket
(),
as_bytes
(
make_span
(
hello_manager
))),
hello_manager
.
size
());
hello_manager
.
size
());
CAF_MESSAGE
(
"wrote "
<<
hello_manager
.
size
()
<<
" bytes."
);
CAF_MESSAGE
(
"wrote "
<<
hello_manager
.
size
()
<<
" bytes."
);
run
();
run
();
CAF_CHECK_EQUAL
(
string_view
(
reinterpret_cast
<
char
*>
(
buf
->
data
()),
CAF_CHECK_EQUAL
(
string_view
(
reinterpret_cast
<
char
*>
(
shared_
buf
->
data
()),
buf
->
size
()),
shared_
buf
->
size
()),
hello_manager
);
hello_manager
);
}
}
CAF_TEST
(
resolve
and
proxy
communication
)
{
CAF_TEST
(
resolve
and
proxy
communication
)
{
using
transport_type
=
stream_transport
<
dummy_application
>
;
using
transport_type
=
stream_transport
<
dummy_application
>
;
std
::
vector
<
byte
>
read_buf
(
1024
);
auto
buf
=
std
::
make_shared
<
std
::
vector
<
byte
>>
();
auto
sockets
=
unbox
(
make_stream_socket_pair
());
nonblocking
(
sockets
.
second
,
true
);
auto
guard
=
detail
::
make_scope_guard
([
&
]
{
close
(
sockets
.
second
);
});
auto
mgr
=
make_endpoint_manager
(
mpx
,
sys
,
auto
mgr
=
make_endpoint_manager
(
mpx
,
sys
,
transport_type
{
sockets
.
first
,
transport_type
{
send_socket_guard
.
release
(),
dummy_application
{
buf
}});
dummy_application
{
shared_buf
}});
CAF_CHECK_EQUAL
(
mgr
->
init
(),
none
);
CAF_CHECK_EQUAL
(
mgr
->
init
(),
none
);
run
();
run
();
mgr
->
resolve
(
unbox
(
make_uri
(
"test:/id/42"
)),
self
);
mgr
->
resolve
(
unbox
(
make_uri
(
"test:/id/42"
)),
self
);
...
@@ -182,13 +194,13 @@ CAF_TEST(resolve and proxy communication) {
...
@@ -182,13 +194,13 @@ CAF_TEST(resolve and proxy communication) {
after
(
std
::
chrono
::
seconds
(
0
))
>>
after
(
std
::
chrono
::
seconds
(
0
))
>>
[
&
]
{
CAF_FAIL
(
"manager did not respond with a proxy."
);
});
[
&
]
{
CAF_FAIL
(
"manager did not respond with a proxy."
);
});
run
();
run
();
auto
read_res
=
read
(
sockets
.
second
,
make_span
(
read
_buf
));
auto
read_res
=
read
(
recv_socket_guard
.
socket
(),
make_span
(
recv
_buf
));
if
(
!
holds_alternative
<
size_t
>
(
read_res
))
if
(
!
holds_alternative
<
size_t
>
(
read_res
))
CAF_FAIL
(
"read() returned an error: "
<<
sys
.
render
(
get
<
sec
>
(
read_res
)));
CAF_FAIL
(
"read() returned an error: "
<<
sys
.
render
(
get
<
sec
>
(
read_res
)));
re
ad
_buf
.
resize
(
get
<
size_t
>
(
read_res
));
re
cv
_buf
.
resize
(
get
<
size_t
>
(
read_res
));
CAF_MESSAGE
(
"receive buffer contains "
<<
re
ad
_buf
.
size
()
<<
" bytes"
);
CAF_MESSAGE
(
"receive buffer contains "
<<
re
cv
_buf
.
size
()
<<
" bytes"
);
message
msg
;
message
msg
;
binary_deserializer
source
{
sys
,
re
ad
_buf
};
binary_deserializer
source
{
sys
,
re
cv
_buf
};
CAF_CHECK_EQUAL
(
source
(
msg
),
none
);
CAF_CHECK_EQUAL
(
source
(
msg
),
none
);
if
(
msg
.
match_elements
<
std
::
string
>
())
if
(
msg
.
match_elements
<
std
::
string
>
())
CAF_CHECK_EQUAL
(
msg
.
get_as
<
std
::
string
>
(
0
),
"hello proxy!"
);
CAF_CHECK_EQUAL
(
msg
.
get_as
<
std
::
string
>
(
0
),
"hello proxy!"
);
...
...
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