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
f582df13
Unverified
Commit
f582df13
authored
Sep 22, 2019
by
Dominik Charousset
Committed by
GitHub
Sep 22, 2019
Browse files
Options
Browse Files
Download
Plain Diff
Merge pull request #26
Add write overload for multiple buffers
parents
47b45d8e
56c98cec
Changes
3
Show whitespace changes
Inline
Side-by-side
Showing
3 changed files
with
127 additions
and
29 deletions
+127
-29
libcaf_net/caf/net/stream_socket.hpp
libcaf_net/caf/net/stream_socket.hpp
+13
-4
libcaf_net/src/stream_socket.cpp
libcaf_net/src/stream_socket.cpp
+45
-0
libcaf_net/test/stream_socket.cpp
libcaf_net/test/stream_socket.cpp
+69
-25
No files found.
libcaf_net/caf/net/stream_socket.hpp
View file @
f582df13
...
@@ -48,21 +48,30 @@ error nodelay(stream_socket x, bool new_value);
...
@@ -48,21 +48,30 @@ error nodelay(stream_socket x, bool new_value);
/// Receives data from `x`.
/// Receives data from `x`.
/// @param x Connected endpoint.
/// @param x Connected endpoint.
/// @param buf Points to destination buffer.
/// @param buf Points to destination buffer.
/// @param buf_size Specifies the maximum size of the buffer in bytes.
/// @returns The number of received bytes on success, an error code otherwise.
/// @returns The number of received bytes on success, an error code otherwise.
/// @relates
pipe
_socket
/// @relates
stream
_socket
/// @post either the result is a `sec` or a positive (non-zero) integer
/// @post either the result is a `sec` or a positive (non-zero) integer
variant
<
size_t
,
sec
>
read
(
stream_socket
x
,
span
<
byte
>
buf
);
variant
<
size_t
,
sec
>
read
(
stream_socket
x
,
span
<
byte
>
buf
);
/// Transmits data from `x` to its peer.
/// Transmits data from `x` to its peer.
/// @param x Connected endpoint.
/// @param x Connected endpoint.
/// @param buf Points to the message to send.
/// @param buf Points to the message to send.
/// @param buf_size Specifies the size of the buffer in bytes.
/// @returns The number of written bytes on success, otherwise an error code.
/// @returns The number of written bytes on success, otherwise an error code.
/// @relates
pipe
_socket
/// @relates
stream
_socket
/// @post either the result is a `sec` or a positive (non-zero) integer
/// @post either the result is a `sec` or a positive (non-zero) integer
variant
<
size_t
,
sec
>
write
(
stream_socket
x
,
span
<
const
byte
>
buf
);
variant
<
size_t
,
sec
>
write
(
stream_socket
x
,
span
<
const
byte
>
buf
);
/// Transmits data from `x` to its peer.
/// @param x Connected endpoint.
/// @param bufs Points to the message to send, scattered across up to 10
/// buffers.
/// @returns The number of written bytes on success, otherwise an error code.
/// @relates stream_socket
/// @post either the result is a `sec` or a positive (non-zero) integer
/// @pre `bufs.size() < 10`
variant
<
size_t
,
sec
>
write
(
stream_socket
x
,
std
::
initializer_list
<
span
<
const
byte
>>
bufs
);
/// Converts the result from I/O operation on a ::stream_socket to either an
/// Converts the result from I/O operation on a ::stream_socket to either an
/// error code or a non-zero positive integer.
/// error code or a non-zero positive integer.
/// @relates stream_socket
/// @relates stream_socket
...
...
libcaf_net/src/stream_socket.cpp
View file @
f582df13
...
@@ -27,6 +27,10 @@
...
@@ -27,6 +27,10 @@
#include "caf/span.hpp"
#include "caf/span.hpp"
#include "caf/variant.hpp"
#include "caf/variant.hpp"
#ifdef CAF_POSIX
# include <sys/uio.h>
#endif
namespace
caf
{
namespace
caf
{
namespace
net
{
namespace
net
{
...
@@ -175,6 +179,47 @@ variant<size_t, sec> write(stream_socket x, span<const byte> buf) {
...
@@ -175,6 +179,47 @@ variant<size_t, sec> write(stream_socket x, span<const byte> buf) {
return
check_stream_socket_io_res
(
res
);
return
check_stream_socket_io_res
(
res
);
}
}
#ifdef CAF_WINDOWS
variant
<
size_t
,
sec
>
write
(
stream_socket
x
,
std
::
initializer_list
<
span
<
const
byte
>>
bufs
)
{
CAF_ASSERT
(
bufs
.
size
()
<
10
);
WSABUF
buf_array
[
10
];
auto
convert
=
[](
span
<
const
byte
>
buf
)
{
auto
data
=
const_cast
<
byte
*>
(
buf
.
data
());
return
WSABUF
{
static_cast
<
ULONG
>
(
buf
.
size
()),
reinterpret_cast
<
CHAR
*>
(
data
)};
};
std
::
transform
(
bufs
.
begin
(),
bufs
.
end
(),
std
::
begin
(
buf_array
),
convert
);
DWORD
bytes_sent
=
0
;
auto
res
=
WSASend
(
x
.
id
,
buf_array
,
static_cast
<
DWORD
>
(
bufs
.
size
()),
&
bytes_sent
,
0
,
nullptr
,
nullptr
);
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
>
(
bytes_sent
);
}
#else // CAF_WINDOWS
variant
<
size_t
,
sec
>
write
(
stream_socket
x
,
std
::
initializer_list
<
span
<
const
byte
>>
bufs
)
{
CAF_ASSERT
(
bufs
.
size
()
<
10
);
iovec
buf_array
[
10
];
auto
convert
=
[](
span
<
const
byte
>
buf
)
{
return
iovec
{
const_cast
<
byte
*>
(
buf
.
data
()),
buf
.
size
()};
};
std
::
transform
(
bufs
.
begin
(),
bufs
.
end
(),
std
::
begin
(
buf_array
),
convert
);
auto
res
=
writev
(
x
.
id
,
buf_array
,
static_cast
<
int
>
(
bufs
.
size
()));
return
check_stream_socket_io_res
(
res
);
}
#endif // CAF_WINDOWS
variant
<
size_t
,
sec
>
variant
<
size_t
,
sec
>
check_stream_socket_io_res
(
std
::
make_signed
<
size_t
>::
type
res
)
{
check_stream_socket_io_res
(
std
::
make_signed
<
size_t
>::
type
res
)
{
if
(
res
==
0
)
if
(
res
==
0
)
...
...
libcaf_net/test/stream_socket.cpp
View file @
f582df13
...
@@ -30,7 +30,15 @@
...
@@ -30,7 +30,15 @@
using
namespace
caf
;
using
namespace
caf
;
using
namespace
caf
::
net
;
using
namespace
caf
::
net
;
CAF_TEST_FIXTURE_SCOPE
(
network_socket_tests
,
host_fixture
)
namespace
{
byte
operator
""
_b
(
unsigned
long
long
x
)
{
return
static_cast
<
byte
>
(
x
);
}
}
// namespace
CAF_TEST_FIXTURE_SCOPE
(
network_socket_simple_tests
,
host_fixture
)
CAF_TEST
(
invalid
socket
)
{
CAF_TEST
(
invalid
socket
)
{
stream_socket
x
;
stream_socket
x
;
...
@@ -39,36 +47,72 @@ CAF_TEST(invalid socket) {
...
@@ -39,36 +47,72 @@ CAF_TEST(invalid socket) {
CAF_CHECK_EQUAL
(
allow_sigpipe
(
x
,
true
),
sec
::
network_syscall_failed
);
CAF_CHECK_EQUAL
(
allow_sigpipe
(
x
,
true
),
sec
::
network_syscall_failed
);
}
}
CAF_TEST
(
connected
socket
pair
)
{
CAF_TEST_FIXTURE_SCOPE_END
()
std
::
vector
<
byte
>
wr_buf
{
byte
(
1
),
byte
(
2
),
byte
(
4
),
byte
(
8
),
byte
(
16
),
byte
(
32
),
byte
(
64
)};
namespace
{
std
::
vector
<
byte
>
rd_buf
(
124
);
CAF_MESSAGE
(
"create sockets and configure nonblocking I/O"
);
struct
fixture
:
host_fixture
{
auto
x
=
unbox
(
make_stream_socket_pair
());
fixture
()
:
rd_buf
(
124
)
{
CAF_CHECK_EQUAL
(
nonblocking
(
x
.
first
,
true
),
caf
::
none
);
std
::
tie
(
first
,
second
)
=
unbox
(
make_stream_socket_pair
());
CAF_CHECK_EQUAL
(
nonblocking
(
x
.
second
,
true
),
caf
::
none
);
CAF_REQUIRE_EQUAL
(
nonblocking
(
first
,
true
),
caf
::
none
);
CAF_CHECK_NOT_EQUAL
(
unbox
(
send_buffer_size
(
x
.
first
)),
0u
);
CAF_REQUIRE_EQUAL
(
nonblocking
(
second
,
true
),
caf
::
none
);
CAF_CHECK_NOT_EQUAL
(
unbox
(
send_buffer_size
(
x
.
second
)),
0u
);
CAF_REQUIRE_NOT_EQUAL
(
unbox
(
send_buffer_size
(
first
)),
0u
);
CAF_MESSAGE
(
"verify nonblocking communication"
);
CAF_REQUIRE_NOT_EQUAL
(
unbox
(
send_buffer_size
(
second
)),
0u
);
CAF_CHECK_EQUAL
(
read
(
x
.
first
,
make_span
(
rd_buf
)),
}
~
fixture
()
{
close
(
first
);
close
(
second
);
}
stream_socket
first
;
stream_socket
second
;
std
::
vector
<
byte
>
rd_buf
;
};
}
// namespace
CAF_TEST_FIXTURE_SCOPE
(
network_socket_tests
,
fixture
)
CAF_TEST
(
read
on
empty
sockets
)
{
CAF_CHECK_EQUAL
(
read
(
first
,
make_span
(
rd_buf
)),
sec
::
unavailable_or_would_block
);
sec
::
unavailable_or_would_block
);
CAF_CHECK_EQUAL
(
read
(
x
.
second
,
make_span
(
rd_buf
)),
CAF_CHECK_EQUAL
(
read
(
second
,
make_span
(
rd_buf
)),
sec
::
unavailable_or_would_block
);
sec
::
unavailable_or_would_block
);
}
CAF_TEST
(
transfer
data
from
first
to
second
socket
)
{
std
::
vector
<
byte
>
wr_buf
{
1
_b
,
2
_b
,
4
_b
,
8
_b
,
16
_b
,
32
_b
,
64
_b
};
CAF_MESSAGE
(
"transfer data from first to second socket"
);
CAF_MESSAGE
(
"transfer data from first to second socket"
);
CAF_CHECK_EQUAL
(
write
(
x
.
first
,
make_span
(
wr_buf
)),
wr_buf
.
size
());
CAF_CHECK_EQUAL
(
write
(
first
,
make_span
(
wr_buf
)),
wr_buf
.
size
());
CAF_CHECK_EQUAL
(
read
(
x
.
second
,
make_span
(
rd_buf
)),
wr_buf
.
size
());
CAF_CHECK_EQUAL
(
read
(
second
,
make_span
(
rd_buf
)),
wr_buf
.
size
());
CAF_CHECK
(
std
::
equal
(
wr_buf
.
begin
(),
wr_buf
.
end
(),
rd_buf
.
begin
()));
CAF_CHECK
(
std
::
equal
(
wr_buf
.
begin
(),
wr_buf
.
end
(),
rd_buf
.
begin
()));
rd_buf
.
assign
(
rd_buf
.
size
(),
byte
(
0
));
rd_buf
.
assign
(
rd_buf
.
size
(),
byte
(
0
));
CAF_MESSAGE
(
"transfer data from second to first socket"
);
}
CAF_CHECK_EQUAL
(
write
(
x
.
second
,
make_span
(
wr_buf
)),
wr_buf
.
size
());
CAF_CHECK_EQUAL
(
read
(
x
.
first
,
make_span
(
rd_buf
)),
wr_buf
.
size
());
CAF_TEST
(
transfer
data
from
second
to
first
socket
)
{
std
::
vector
<
byte
>
wr_buf
{
1
_b
,
2
_b
,
4
_b
,
8
_b
,
16
_b
,
32
_b
,
64
_b
};
CAF_CHECK_EQUAL
(
write
(
second
,
make_span
(
wr_buf
)),
wr_buf
.
size
());
CAF_CHECK_EQUAL
(
read
(
first
,
make_span
(
rd_buf
)),
wr_buf
.
size
());
CAF_CHECK
(
std
::
equal
(
wr_buf
.
begin
(),
wr_buf
.
end
(),
rd_buf
.
begin
()));
CAF_CHECK
(
std
::
equal
(
wr_buf
.
begin
(),
wr_buf
.
end
(),
rd_buf
.
begin
()));
rd_buf
.
assign
(
rd_buf
.
size
(),
byte
(
0
));
}
CAF_MESSAGE
(
"shut down first socket and observe shutdown on the second one"
);
close
(
x
.
first
);
CAF_TEST
(
shut
down
first
socket
and
observe
shutdown
on
the
second
one
)
{
CAF_CHECK_EQUAL
(
read
(
x
.
second
,
make_span
(
rd_buf
)),
sec
::
socket_disconnected
);
close
(
first
);
CAF_MESSAGE
(
"done (cleanup)"
);
CAF_CHECK_EQUAL
(
read
(
second
,
make_span
(
rd_buf
)),
sec
::
socket_disconnected
);
close
(
x
.
second
);
first
.
id
=
invalid_socket_id
;
}
CAF_TEST
(
transfer
data
using
multiple
buffers
)
{
std
::
vector
<
byte
>
wr_buf_1
{
1
_b
,
2
_b
,
4
_b
};
std
::
vector
<
byte
>
wr_buf_2
{
8
_b
,
16
_b
,
32
_b
,
64
_b
};
std
::
vector
<
byte
>
full_buf
;
full_buf
.
insert
(
full_buf
.
end
(),
wr_buf_1
.
begin
(),
wr_buf_1
.
end
());
full_buf
.
insert
(
full_buf
.
end
(),
wr_buf_2
.
begin
(),
wr_buf_2
.
end
());
CAF_CHECK_EQUAL
(
write
(
second
,
{
make_span
(
wr_buf_1
),
make_span
(
wr_buf_2
)}),
full_buf
.
size
());
CAF_CHECK_EQUAL
(
read
(
first
,
make_span
(
rd_buf
)),
full_buf
.
size
());
CAF_CHECK
(
std
::
equal
(
full_buf
.
begin
(),
full_buf
.
end
(),
rd_buf
.
begin
()));
}
}
CAF_TEST_FIXTURE_SCOPE_END
()
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