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
042b07c3
Commit
042b07c3
authored
Mar 13, 2021
by
Dominik Charousset
Browse files
Options
Browse Files
Download
Plain Diff
Merge branch topic/neverlord/length-prefix-framing
parents
5b3dbf8f
827a1952
Changes
4
Hide whitespace changes
Inline
Side-by-side
Showing
4 changed files
with
286 additions
and
229 deletions
+286
-229
libcaf_net/caf/net/length_prefix_framing.hpp
libcaf_net/caf/net/length_prefix_framing.hpp
+121
-77
libcaf_net/caf/net/message_oriented_layer_ptr.hpp
libcaf_net/caf/net/message_oriented_layer_ptr.hpp
+87
-0
libcaf_net/caf/net/mixed_message_oriented_layer_ptr.hpp
libcaf_net/caf/net/mixed_message_oriented_layer_ptr.hpp
+0
-1
libcaf_net/test/net/length_prefix_framing.cpp
libcaf_net/test/net/length_prefix_framing.cpp
+78
-151
No files found.
libcaf_net/caf/net/length_prefix_framing.hpp
View file @
042b07c3
...
...
@@ -12,6 +12,7 @@
#include "caf/byte_span.hpp"
#include "caf/detail/network_order.hpp"
#include "caf/error.hpp"
#include "caf/net/message_oriented_layer_ptr.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/sec.hpp"
#include "caf/span.hpp"
...
...
@@ -34,63 +35,26 @@ public:
using
length_prefix_type
=
uint32_t
;
static
constexpr
size_t
max_message_length
=
INT32_MAX
;
static
constexpr
size_t
max_message_length
=
INT32_MAX
-
sizeof
(
uint32_t
)
;
// -- interface for the upper layer ------------------------------------------
template
<
class
LowerLayer
>
class
access
{
public:
access
(
LowerLayer
*
lower_layer
,
length_prefix_framing
*
this_layer
)
:
lower_layer_
(
lower_layer
),
this_layer_
(
this_layer
)
{
// nop
}
void
begin_message
()
{
lower_layer_
->
begin_output
();
auto
&
buf
=
message_buffer
();
message_offset_
=
buf
.
size
();
buf
.
insert
(
buf
.
end
(),
4
,
byte
{
0
});
}
byte_buffer
&
message_buffer
()
{
return
lower_layer_
->
output_buffer
();
}
bool
end_message
()
{
using
detail
::
to_network_order
;
auto
&
buf
=
message_buffer
();
auto
msg_begin
=
buf
.
begin
()
+
message_offset_
;
auto
msg_size
=
std
::
distance
(
msg_begin
+
4
,
buf
.
end
());
if
(
msg_size
>
0
&&
msg_size
<
max_message_length
)
{
auto
u32_size
=
to_network_order
(
static_cast
<
uint32_t
>
(
msg_size
));
memcpy
(
std
::
addressof
(
*
msg_begin
),
&
u32_size
,
4
);
return
true
;
}
else
{
abort_reason
(
make_error
(
sec
::
runtime_error
,
msg_size
==
0
?
"logic error: message of size 0"
:
"maximum message size exceeded"
));
return
false
;
}
}
static
constexpr
uint32_t
default_receive_size
=
4
*
1024
;
// 4kb.
bool
can_send_more
()
const
noexcept
{
return
lower_layer_
->
can_send_more
();
}
// -- constructors, destructors, and assignment operators --------------------
void
abort_reason
(
error
reason
)
{
return
lower_layer_
->
abort_reason
(
std
::
move
(
reason
));
}
template
<
class
...
Ts
>
explicit
length_prefix_framing
(
Ts
&&
...
xs
)
:
upper_layer_
(
std
::
forward
<
Ts
>
(
xs
)...)
{
// nop
}
void
configure_read
(
receive_policy
policy
)
{
lower_layer_
->
configure_read
(
policy
);
}
// -- initialization ---------------------------------------------------------
private:
LowerLayer
*
lower_layer_
;
length_prefix_framing
*
this_layer_
;
size_t
message_offset_
=
0
;
};
template
<
class
LowerLayerPtr
>
error
init
(
socket_manager
*
owner
,
LowerLayerPtr
down
,
const
settings
&
cfg
)
{
down
->
configure_read
(
receive_policy
::
between
(
sizeof
(
uint32_t
),
default_receive_size
));
return
upper_layer_
.
init
(
owner
,
this_layer_ptr
(
down
),
cfg
);
}
// -- properties -------------------------------------------------------------
...
...
@@ -101,42 +65,122 @@ public:
const
auto
&
upper_layer
()
const
noexcept
{
return
upper_layer_
;
}
// -- role: upper layer ------------------------------------------------------
template
<
class
LowerLayer
>
bool
prepare_send
(
LowerLayer
&
down
)
{
access
<
LowerLayer
>
this_layer
{
&
down
,
this
};
return
upper_layer_
.
prepare_send
(
this_layer
);
// -- interface for the upper layer ------------------------------------------
template
<
class
LowerLayerPtr
>
void
begin_message
(
LowerLayerPtr
down
)
{
down
->
begin_output
();
auto
&
buf
=
down
->
output_buffer
();
message_offset_
=
buf
.
size
();
buf
.
insert
(
buf
.
end
(),
4
,
byte
{
0
});
}
template
<
class
LowerLayerPtr
>
byte_buffer
&
message_buffer
(
LowerLayerPtr
down
)
{
return
down
->
output_buffer
();
}
template
<
class
LowerLayerPtr
>
[[
nodiscard
]]
bool
end_message
(
LowerLayerPtr
down
)
{
using
detail
::
to_network_order
;
auto
&
buf
=
down
->
output_buffer
();
CAF_ASSERT
(
message_offset_
<
buf
.
size
());
auto
msg_begin
=
buf
.
begin
()
+
message_offset_
;
auto
msg_size
=
std
::
distance
(
msg_begin
+
4
,
buf
.
end
());
if
(
msg_size
>
0
&&
static_cast
<
size_t
>
(
msg_size
)
<
max_message_length
)
{
auto
u32_size
=
to_network_order
(
static_cast
<
uint32_t
>
(
msg_size
));
memcpy
(
std
::
addressof
(
*
msg_begin
),
&
u32_size
,
4
);
return
true
;
}
else
{
auto
err
=
make_error
(
sec
::
runtime_error
,
msg_size
==
0
?
"logic error: message of size 0"
:
"maximum message size exceeded"
);
down
->
abort_reason
(
std
::
move
(
err
));
return
false
;
}
}
// -- interface for the lower layer ------------------------------------------
template
<
class
LowerLayerPtr
>
bool
prepare_send
(
LowerLayerPtr
down
)
{
return
upper_layer_
.
prepare_send
(
this_layer_ptr
(
down
));
}
template
<
class
LowerLayerPtr
>
bool
done_sending
(
LowerLayerPtr
down
)
{
return
upper_layer_
.
done_sending
(
this_layer_ptr
(
down
));
}
template
<
class
LowerLayer
>
bool
done_sending
(
LowerLayer
&
down
)
{
access
<
LowerLayer
>
this_layer
{
&
down
,
this
};
return
upper_layer_
.
done_sending
(
this_layer
);
template
<
class
LowerLayerPtr
>
void
abort
(
LowerLayerPtr
down
,
const
error
&
reason
)
{
upper_layer_
.
abort
(
this_layer_ptr
(
down
),
reason
);
}
template
<
class
LowerLayer
>
void
abort
(
LowerLayer
&
down
,
const
error
&
reason
)
{
access
<
LowerLayer
>
this_layer
{
&
down
,
this
};
return
upper_layer_
.
abort
(
this_layer
,
reason
);
template
<
class
LowerLayerPtr
>
ptrdiff_t
consume
(
LowerLayerPtr
down
,
byte_span
input
,
byte_span
)
{
auto
buffer
=
input
;
auto
consumed
=
ptrdiff_t
{
0
};
auto
this_layer
=
this_layer_ptr
(
down
);
for
(;;)
{
if
(
input
.
size
()
<
sizeof
(
uint32_t
))
{
return
consumed
;
}
else
{
auto
[
msg_size
,
sub_buffer
]
=
split
(
input
);
if
(
msg_size
==
0
)
{
consumed
+=
static_cast
<
ptrdiff_t
>
(
sizeof
(
uint32_t
));
input
=
sub_buffer
;
}
else
if
(
msg_size
>
max_message_length
)
{
auto
err
=
make_error
(
sec
::
runtime_error
,
"maximum message size exceeded"
);
down
->
abort_reason
(
std
::
move
(
err
));
return
-
1
;
}
else
if
(
msg_size
>
sub_buffer
.
size
())
{
if
(
msg_size
+
sizeof
(
uint32_t
)
>
receive_buf_upper_bound_
)
{
auto
min_read_size
=
static_cast
<
uint32_t
>
(
sizeof
(
uint32_t
));
receive_buf_upper_bound_
=
static_cast
<
uint32_t
>
(
msg_size
+
sizeof
(
uint32_t
));
down
->
configure_read
(
receive_policy
::
between
(
min_read_size
,
receive_buf_upper_bound_
));
}
return
consumed
;
}
else
{
auto
msg
=
sub_buffer
.
subspan
(
0
,
msg_size
);
if
(
auto
res
=
upper_layer_
.
consume
(
this_layer
,
msg
);
res
>=
0
)
{
consumed
+=
static_cast
<
ptrdiff_t
>
(
msg
.
size
())
+
sizeof
(
uint32_t
);
input
=
sub_buffer
.
subspan
(
msg_size
);
}
else
{
return
-
1
;
}
}
}
}
}
template
<
class
LowerLayer
>
ptrdiff_t
consume
(
LowerLayer
&
down
,
byte_span
buffer
,
byte_span
)
{
using
detail
::
from_network_order
;
if
(
buffer
.
size
()
<
4
)
return
0
;
uint32_t
u32_size
=
0
;
memcpy
(
&
u32_size
,
buffer
.
data
(),
4
);
auto
msg_size
=
static_cast
<
size_t
>
(
from_network_order
(
u32_size
));
if
(
buffer
.
size
()
<
msg_size
+
4
)
return
0
;
upper_layer_
.
consume
(
down
,
make_span
(
buffer
.
data
()
+
4
,
msg_size
));
return
msg_size
+
4
;
// -- convenience functions --------------------------------------------------
static
std
::
pair
<
size_t
,
byte_span
>
split
(
byte_span
buffer
)
noexcept
{
CAF_ASSERT
(
buffer
.
size
()
>=
sizeof
(
uint32_t
));
auto
u32_size
=
uint32_t
{
0
};
memcpy
(
&
u32_size
,
buffer
.
data
(),
sizeof
(
uint32_t
));
auto
msg_size
=
static_cast
<
size_t
>
(
detail
::
from_network_order
(
u32_size
));
return
std
::
make_pair
(
msg_size
,
buffer
.
subspan
(
sizeof
(
uint32_t
)));
}
private:
// -- implementation details -------------------------------------------------
template
<
class
LowerLayerPtr
>
auto
this_layer_ptr
(
LowerLayerPtr
down
)
{
return
make_message_oriented_layer_ptr
(
this
,
down
);
}
// -- member variables -------------------------------------------------------
UpperLayer
upper_layer_
;
size_t
message_offset_
=
0
;
uint32_t
receive_buf_upper_bound_
=
default_receive_size
;
};
}
// namespace caf::net
libcaf_net/caf/net/message_oriented_layer_ptr.hpp
0 → 100644
View file @
042b07c3
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
namespace
caf
::
net
{
/// Wraps a pointer to a mixed-message-oriented layer with a pointer to its
/// lower layer. Both pointers are then used to implement the interface required
/// for a mixed-message-oriented layer when calling into its upper layer.
template
<
class
Layer
,
class
LowerLayerPtr
>
class
message_oriented_layer_ptr
{
public:
class
access
{
public:
access
(
Layer
*
layer
,
LowerLayerPtr
down
)
:
lptr_
(
layer
),
llptr_
(
down
)
{
// nop
}
bool
can_send_more
()
const
noexcept
{
return
lptr_
->
can_send_more
(
llptr_
);
}
auto
handle
()
const
noexcept
{
return
lptr_
->
handle
(
llptr_
);
}
void
begin_message
()
{
lptr_
->
begin_message
(
llptr_
);
}
[[
nodiscard
]]
auto
&
message_buffer
()
{
return
lptr_
->
message_buffer
(
llptr_
);
}
[[
nodiscard
]]
bool
end_message
()
{
return
lptr_
->
end_message
(
llptr_
);
}
void
abort_reason
(
error
reason
)
{
return
lptr_
->
abort_reason
(
llptr_
,
std
::
move
(
reason
));
}
const
error
&
abort_reason
()
{
return
lptr_
->
abort_reason
(
llptr_
);
}
private:
Layer
*
lptr_
;
LowerLayerPtr
llptr_
;
};
message_oriented_layer_ptr
(
Layer
*
layer
,
LowerLayerPtr
down
)
:
access_
(
layer
,
down
)
{
// nop
}
message_oriented_layer_ptr
(
const
message_oriented_layer_ptr
&
)
=
default
;
explicit
operator
bool
()
const
noexcept
{
return
true
;
}
access
*
operator
->
()
const
noexcept
{
return
&
access_
;
}
access
&
operator
*
()
const
noexcept
{
return
access_
;
}
private:
mutable
access
access_
;
};
template
<
class
Layer
,
class
LowerLayerPtr
>
auto
make_message_oriented_layer_ptr
(
Layer
*
this_layer
,
LowerLayerPtr
down
)
{
using
result_t
=
message_oriented_layer_ptr
<
Layer
,
LowerLayerPtr
>
;
return
result_t
{
this_layer
,
down
};
}
}
// namespace caf::net
libcaf_net/caf/net/mixed_message_oriented_layer_ptr.hpp
View file @
042b07c3
...
...
@@ -7,7 +7,6 @@
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/receive_policy.hpp"
namespace
caf
::
net
{
...
...
libcaf_net/test/net/length_prefix_framing.cpp
View file @
042b07c3
...
...
@@ -6,9 +6,9 @@
#include "caf/net/length_prefix_framing.hpp"
#include "
caf/test/dsl
.hpp"
#include "
net-test
.hpp"
#include <
dequ
e>
#include <
cctyp
e>
#include <numeric>
#include <vector>
...
...
@@ -21,179 +21,106 @@
#include "caf/tag/message_oriented.hpp"
using
namespace
caf
;
using
namespace
std
::
literals
;
namespace
{
/// upper layer: expect messages
/// Needs to be initilized by the layer two steps down.
struct
ul_expect_messages
{
using
string_list
=
std
::
vector
<
std
::
string
>
;
struct
app
{
using
input_tag
=
tag
::
message_oriented
;
void
set_expected_messages
(
std
::
vector
<
byte_buffer
>
messages
)
{
expected_messages
.
clear
();
for
(
auto
&
msg
:
messages
)
expected_messages
.
emplace_back
(
std
::
move
(
msg
));
template
<
class
LowerLayerPtr
>
caf
::
error
init
(
net
::
socket_manager
*
,
LowerLayerPtr
,
const
settings
&
)
{
return
none
;
}
void
add_expected_messages
(
std
::
vector
<
byte_buffer
>
messages
)
{
for
(
auto
&
msg
:
messages
)
expected_messages
.
emplace_back
(
std
::
move
(
msg
))
;
template
<
class
LowerLayerPtr
>
bool
prepare_send
(
LowerLayerPtr
)
{
return
true
;
}
template
<
class
LowerLayer
>
ptrdiff_t
consume
(
LowerLayer
&
,
byte_span
buffer
)
{
CAF_REQUIRE
(
expected_messages
.
size
()
>
0
);
auto
&
next
=
expected_messages
.
front
();
CAF_CHECK_EQUAL
(
next
.
size
(),
buffer
.
size
());
CAF_CHECK
(
std
::
equal
(
next
.
begin
(),
next
.
end
(),
buffer
.
begin
()));
expected_messages
.
pop_front
();
return
buffer
.
size
();
template
<
class
LowerLayerPtr
>
bool
done_sending
(
LowerLayerPtr
)
{
return
true
;
}
std
::
deque
<
byte_buffer
>
expected_messages
;
};
/// lower layer: offer stream for message parsing
template
<
class
UpperLayer
>
struct
ll_provide_stream_for_messages
{
using
output_tag
=
tag
::
stream_oriented
;
void
set_expectations
(
std
::
vector
<
byte
>
data
,
std
::
vector
<
byte_buffer
>
messages
)
{
data_stream
=
std
::
move
(
data
);
auto
&
checking_layer
=
upper_layer
.
upper_layer
();
checking_layer
.
set_expected_messages
(
messages
);
template
<
class
LowerLayerPtr
>
void
abort
(
LowerLayerPtr
,
const
error
&
)
{
// nop
}
void
add_expectations
(
const
std
::
vector
<
byte
>&
data
,
std
::
vector
<
byte_buffer
>
messages
)
{
data_stream
.
insert
(
data_stream
.
end
(),
data
.
begin
(),
data
.
end
());
auto
&
checking_layer
=
upper_layer
.
upper_layer
();
checking_layer
.
add_expect_messages
(
messages
);
}
void
run
()
{
CAF_CHECK
(
!
data_stream
.
empty
());
while
(
processed
!=
data_stream
.
size
())
{
auto
all_data
=
make_span
(
data_stream
.
data
()
+
processed
,
data_stream
.
size
()
-
processed
);
auto
new_data
=
make_span
(
data_stream
.
data
()
+
offered
,
data_stream
.
size
()
-
offered
);
auto
newly_offered
=
new_data
.
size
();
auto
consumed
=
upper_layer
.
consume
(
*
this
,
all_data
,
new_data
);
CAF_CHECK
(
consumed
>=
0
);
CAF_CHECK
(
static_cast
<
size_t
>
(
consumed
)
<=
data_stream
.
size
());
offered
+=
newly_offered
;
processed
+=
consumed
;
if
(
consumed
>
0
)
{
data_stream
.
erase
(
data_stream
.
begin
(),
data_stream
.
begin
()
+
consumed
);
offered
-=
processed
;
processed
=
0
;
}
if
(
consumed
==
0
||
data_stream
.
empty
())
return
;
template
<
class
LowerLayerPtr
>
ptrdiff_t
consume
(
LowerLayerPtr
down
,
byte_span
buf
)
{
auto
printable
=
[](
byte
x
)
{
return
::
isprint
(
static_cast
<
uint8_t
>
(
x
));
};
if
(
CHECK
(
std
::
all_of
(
buf
.
begin
(),
buf
.
end
(),
printable
)))
{
auto
str_buf
=
reinterpret_cast
<
char
*>
(
buf
.
data
());
inputs
.
emplace_back
(
std
::
string
{
str_buf
,
buf
.
size
()});
std
::
string
response
=
"ok "
;
response
+=
std
::
to_string
(
inputs
.
size
());
auto
response_bytes
=
as_bytes
(
make_span
(
response
));
down
->
begin_message
();
auto
&
buf
=
down
->
message_buffer
();
buf
.
insert
(
buf
.
end
(),
response_bytes
.
begin
(),
response_bytes
.
end
());
CHECK
(
down
->
end_message
());
return
static_cast
<
ptrdiff_t
>
(
buf
.
size
());
}
else
{
return
-
1
;
}
}
size_t
processed
=
0
;
size_t
offered
=
0
;
std
::
vector
<
byte
>
data_stream
;
UpperLayer
upper_layer
;
std
::
vector
<
std
::
string
>
inputs
;
};
template
<
class
...
Ts
>
byte_buffer
to_buf
(
const
Ts
&
...
xs
)
{
byte_buffer
buf
;
binary_serializer
sink
{
nullptr
,
buf
};
if
(
!
(
sink
.
apply
(
xs
)
&&
...))
CAF_FAIL
(
"to_buf failed: "
<<
sink
.
get_error
());
return
buf
;
}
void
encode_message
(
std
::
vector
<
byte
>&
data
,
const
byte_buffer
&
message
)
{
void
encode
(
byte_buffer
&
buf
,
string_view
msg
)
{
using
detail
::
to_network_order
;
auto
current_size
=
data
.
size
();
data
.
insert
(
data
.
end
(),
4
,
byte
{
0
});
auto
msg_begin
=
data
.
begin
()
+
current_size
;
auto
msg_size
=
message
.
size
();
auto
u32_size
=
to_network_order
(
static_cast
<
uint32_t
>
(
msg_size
));
memcpy
(
std
::
addressof
(
*
msg_begin
),
&
u32_size
,
4
);
data
.
insert
(
data
.
end
(),
message
.
begin
(),
message
.
end
());
auto
prefix
=
to_network_order
(
static_cast
<
uint32_t
>
(
msg
.
size
()));
auto
prefix_bytes
=
as_bytes
(
make_span
(
&
prefix
,
1
));
buf
.
insert
(
buf
.
end
(),
prefix_bytes
.
begin
(),
prefix_bytes
.
end
());
auto
bytes
=
as_bytes
(
make_span
(
msg
));
buf
.
insert
(
buf
.
end
(),
bytes
.
begin
(),
bytes
.
end
());
}
struct
fixture
{
using
test_layers
=
ll_provide_stream_for_messages
<
net
::
length_prefix_framing
<
ul_expect_messages
>>
;
void
generate_messages
(
size_t
num
,
size_t
factor
=
10
)
{
for
(
size_t
n
=
1
;
n
<=
num
;
n
+=
1
)
{
std
::
vector
<
int
>
buf
(
n
*
factor
);
std
::
iota
(
buf
.
begin
(),
buf
.
end
(),
n
);
messages
.
emplace_back
(
to_buf
(
buf
));
auto
decode
(
byte_buffer
&
buf
)
{
auto
printable
=
[](
byte
x
)
{
return
::
isprint
(
static_cast
<
uint8_t
>
(
x
));
};
string_list
result
;
auto
input
=
make_span
(
buf
);
while
(
!
input
.
empty
())
{
auto
[
msg_size
,
msg
]
=
net
::
length_prefix_framing
<
app
>::
split
(
input
);
if
(
msg_size
>
msg
.
size
())
{
CAF_FAIL
(
"cannot decode buffer: invalid message size"
);
}
else
if
(
!
std
::
all_of
(
msg
.
begin
(),
msg
.
begin
()
+
msg_size
,
printable
))
{
CAF_FAIL
(
"cannot decode buffer: unprintable characters found in message"
);
}
else
{
auto
str
=
std
::
string
{
reinterpret_cast
<
char
*>
(
msg
.
data
()),
msg_size
};
result
.
emplace_back
(
std
::
move
(
str
));
input
=
msg
.
subspan
(
msg_size
);
}
for
(
auto
&
msg
:
messages
)
encode_message
(
data
,
msg
);
}
void
set_expectations
()
{
layers
.
set_expectations
(
data
,
messages
);
}
void
test_receive_data
()
{
layers
.
run
();
}
void
clear
()
{
data
.
clear
();
messages
.
clear
();
}
test_layers
layers
;
std
::
vector
<
byte
>
data
;
std
::
vector
<
byte_buffer
>
messages
;
};
return
result
;
}
}
// namespace
CAF_TEST_FIXTURE_SCOPE
(
length_prefix_framing_tests
,
fixture
)
CAF_TEST
(
process
messages
)
{
// Single message.
generate_messages
(
1
);
set_expectations
();
test_receive_data
();
clear
();
// Multiple messages.
generate_messages
(
10
);
set_expectations
();
test_receive_data
();
}
CAF_TEST
(
incomplete
message
)
{
generate_messages
(
1
,
1000
);
CAF_MESSAGE
(
"data.size() = "
<<
data
.
size
());
auto
initial_size
=
data
.
size
();
auto
data_copy
=
data
;
auto
mid
=
data
.
size
()
/
2
;
data
.
resize
(
mid
);
CAF_MESSAGE
(
"data.size() = "
<<
data
.
size
());
data_copy
.
erase
(
data_copy
.
begin
(),
data_copy
.
begin
()
+
mid
);
CAF_MESSAGE
(
"data_copy.size() = "
<<
data_copy
.
size
());
CAF_REQUIRE
(
data
.
size
()
+
data_copy
.
size
()
==
initial_size
);
// Don't set expectations because there shouldn't be a complete message
// in the bytes.
auto
messages_copy
=
messages
;
messages
.
clear
();
CAF_REQUIRE
(
messages
.
empty
());
set_expectations
();
test_receive_data
();
data
.
insert
(
data
.
end
(),
data_copy
.
begin
(),
data_copy
.
end
());
messages
=
messages_copy
;
set_expectations
();
test_receive_data
();
SCENARIO
(
"length-prefix framing reads data with 32-bit size headers"
)
{
GIVEN
(
"a length_prefix_framing with an app that consumed strings"
)
{
mock_stream_transport
<
net
::
length_prefix_framing
<
app
>>
uut
;
CHECK_EQ
(
uut
.
init
(),
error
{});
WHEN
(
"pushing data into the unit-under-test"
)
{
encode
(
uut
.
input
,
"hello"
);
encode
(
uut
.
input
,
"world"
);
auto
input_size
=
static_cast
<
ptrdiff_t
>
(
uut
.
input
.
size
());
CHECK_EQ
(
uut
.
handle_input
(),
input_size
);
THEN
(
"the app receives all strings as individual messages"
)
{
auto
&
state
=
uut
.
upper_layer
.
upper_layer
();
if
(
CHECK_EQ
(
state
.
inputs
.
size
(),
2u
))
{
CHECK_EQ
(
state
.
inputs
[
0
],
"hello"
);
CHECK_EQ
(
state
.
inputs
[
1
],
"world"
);
}
auto
response
=
string_view
{
reinterpret_cast
<
char
*>
(
uut
.
output
.
data
()),
uut
.
output
.
size
()};
CHECK_EQ
(
decode
(
uut
.
output
),
string_list
({
"ok 1"
,
"ok 2"
}));
}
}
}
}
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