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
68451a12
Commit
68451a12
authored
Aug 05, 2019
by
Dominik Charousset
Committed by
GitHub
Aug 05, 2019
Browse files
Options
Browse Files
Download
Plain Diff
Merge pull request #9
Add scribe transport-policy
parents
b829c8d9
14788d72
Changes
7
Show whitespace changes
Inline
Side-by-side
Showing
7 changed files
with
781 additions
and
0 deletions
+781
-0
libcaf_net/CMakeLists.txt
libcaf_net/CMakeLists.txt
+1
-0
libcaf_net/caf/net/dispatcher.hpp
libcaf_net/caf/net/dispatcher.hpp
+92
-0
libcaf_net/caf/net/receive_policy.hpp
libcaf_net/caf/net/receive_policy.hpp
+61
-0
libcaf_net/caf/policy/scribe.hpp
libcaf_net/caf/policy/scribe.hpp
+149
-0
libcaf_net/src/scribe.cpp
libcaf_net/src/scribe.cpp
+77
-0
libcaf_net/test/scribe.cpp
libcaf_net/test/scribe.cpp
+181
-0
libcaf_net/test/string_application.cpp
libcaf_net/test/string_application.cpp
+220
-0
No files found.
libcaf_net/CMakeLists.txt
View file @
68451a12
...
...
@@ -18,6 +18,7 @@ set(LIBCAF_NET_SRCS
src/socket.cpp
src/socket_manager.cpp
src/stream_socket.cpp
src/scribe.cpp
)
add_custom_target
(
libcaf_net
)
...
...
libcaf_net/caf/net/dispatcher.hpp
0 → 100644
View file @
68451a12
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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
namespace
caf
{
namespace
net
{
/// Implements the interface for transport and application policies and
/// dispatches member functions either to `decorator` or `fallback`.
template
<
class
Decorator
,
class
Fallback
>
class
dispatcher
{
public:
dispatcher
(
Decorator
&
decorator
,
Fallback
&
fallback
)
:
decorator_
(
decorator
),
fallback_
(
fallback
)
{
// nop
}
template
<
class
Header
>
void
write_packet
(
const
Header
&
header
,
span
<
const
byte
>
payload
)
{
decltype
(
write_packet_delegate_test
<
Decorator
>
(
nullptr
,
header
,
payload
))
x
;
write_packet_delegate
(
x
,
header
,
payload
);
}
actor_system
&
system
()
{
fallback_
.
system
();
}
void
cancel_timeout
(
atom_value
type
,
uint64_t
id
)
{
fallback_
.
cancel_timeout
(
type
,
id
);
}
void
set_timeout
(
timestamp
tout
,
atom_value
type
,
uint64_t
id
)
{
fallback_
.
set_timeout
(
tout
,
type
,
id
);
}
typename
Fallback
::
transport_type
&
transport
()
{
return
fallback_
.
transport_
;
}
typename
Fallback
::
application_type
&
application
()
{
return
fallback_
.
application_
;
}
private:
template
<
class
U
,
class
Header
>
static
auto
write_packet_delegate_test
(
U
*
x
,
const
Header
&
hdr
,
span
<
const
byte
>
payload
)
->
decltype
(
x
->
write_packet
(
hdr
,
payload
),
std
::
true_type
());
template
<
class
U
>
static
auto
write_packet_delegate_test
(...)
->
std
::
false_type
;
template
<
class
Header
>
void
write_packet_delegate
(
std
::
true_type
,
const
Header
&
header
,
span
<
const
byte
>
payload
)
{
decorator_
.
write_packet
(
header
,
payload
);
}
template
<
class
Header
>
void
write_packet_delegate
(
std
::
false_type
,
const
Header
&
header
,
span
<
const
byte
>
payload
)
{
fallback_
.
write_packet
(
header
,
payload
);
}
Object
&
decorator_
;
Fallback
&
fallback_
;
};
template
<
class
Decorator
,
class
Fallback
>
dispatcher
<
Decorator
,
Fallback
>
make_dispatcher
(
Decorator
&
decorator
,
Fallback
&
fallback
)
{
return
{
decorator
,
fallback
};
}
}
// namespace net
}
// namespace caf
libcaf_net/caf/net/receive_policy.hpp
0 → 100644
View file @
68451a12
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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 <cstddef>
#include <string>
#include <utility>
#include "caf/config.hpp"
namespace
caf
{
namespace
net
{
enum
class
receive_policy_flag
:
unsigned
{
at_least
,
at_most
,
exactly
};
inline
std
::
string
to_string
(
receive_policy_flag
x
)
{
return
x
==
receive_policy_flag
::
at_least
?
"at_least"
:
(
x
==
receive_policy_flag
::
at_most
?
"at_most"
:
"exactly"
);
}
class
receive_policy
{
public:
receive_policy
()
=
delete
;
using
config
=
std
::
pair
<
receive_policy_flag
,
size_t
>
;
static
config
at_least
(
size_t
num_bytes
)
{
CAF_ASSERT
(
num_bytes
>
0
);
return
{
receive_policy_flag
::
at_least
,
num_bytes
};
}
static
config
at_most
(
size_t
num_bytes
)
{
CAF_ASSERT
(
num_bytes
>
0
);
return
{
receive_policy_flag
::
at_most
,
num_bytes
};
}
static
config
exactly
(
size_t
num_bytes
)
{
CAF_ASSERT
(
num_bytes
>
0
);
return
{
receive_policy_flag
::
exactly
,
num_bytes
};
}
};
}
// namespace net
}
// namespace caf
libcaf_net/caf/policy/scribe.hpp
0 → 100644
View file @
68451a12
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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/error.hpp>
#include <caf/fwd.hpp>
#include <caf/logger.hpp>
#include <caf/net/endpoint_manager.hpp>
#include <caf/net/receive_policy.hpp>
#include <caf/net/stream_socket.hpp>
#include <caf/sec.hpp>
#include <caf/span.hpp>
#include <caf/variant.hpp>
namespace
caf
{
namespace
policy
{
/// Implements a scribe policy that manages a stream socket.
class
scribe
{
public:
explicit
scribe
(
net
::
stream_socket
handle
);
net
::
stream_socket
handle
()
const
noexcept
{
return
handle_
;
}
template
<
class
Parent
>
error
init
(
Parent
&
parent
)
{
parent
.
application
().
init
(
parent
);
parent
.
mask_add
(
net
::
operation
::
read
);
return
none
;
}
template
<
class
Parent
>
bool
handle_read_event
(
Parent
&
parent
)
{
void
*
buf
=
read_buf_
.
data
()
+
collected_
;
size_t
len
=
read_threshold_
-
collected_
;
CAF_LOG_TRACE
(
CAF_ARG
(
handle_
.
id
)
<<
CAF_ARG
(
len
));
auto
ret
=
read
(
handle_
,
buf
,
len
);
// Update state.
if
(
auto
num_bytes
=
get_if
<
size_t
>
(
&
ret
))
{
CAF_LOG_DEBUG
(
CAF_ARG
(
len
)
<<
CAF_ARG
(
handle_
.
id
)
<<
CAF_ARG
(
*
num_bytes
));
collected_
+=
*
num_bytes
;
if
(
collected_
>=
read_threshold_
)
{
parent
.
application
().
handle_data
(
*
this
,
read_buf_
);
prepare_next_read
();
}
return
true
;
}
else
{
auto
err
=
get
<
sec
>
(
ret
);
CAF_LOG_DEBUG
(
"receive failed"
<<
CAF_ARG
(
err
));
parent
.
application
().
handle_error
(
err
);
return
false
;
}
}
template
<
class
Parent
>
bool
handle_write_event
(
Parent
&
parent
)
{
// Try to write leftover data.
write_some
(
parent
);
// Get new data from parent.
for
(
auto
msg
=
parent
.
next_message
();
msg
!=
nullptr
;
msg
=
parent
.
next_message
())
{
parent
.
application
().
write_message
(
*
this
,
std
::
move
(
msg
));
}
// Write prepared data.
return
write_some
(
parent
);
}
template
<
class
Parent
>
bool
write_some
(
Parent
&
parent
)
{
if
(
write_buf_
.
empty
())
return
false
;
auto
len
=
write_buf_
.
size
()
-
written_
;
void
*
buf
=
write_buf_
.
data
()
+
written_
;
CAF_LOG_TRACE
(
CAF_ARG
(
handle_
.
id
)
<<
CAF_ARG
(
len
));
auto
ret
=
net
::
write
(
handle_
,
buf
,
len
);
if
(
auto
num_bytes
=
get_if
<
size_t
>
(
&
ret
))
{
CAF_LOG_DEBUG
(
CAF_ARG
(
len
)
<<
CAF_ARG
(
handle_
.
id
)
<<
CAF_ARG
(
*
num_bytes
));
// Update state.
written_
+=
*
num_bytes
;
if
(
written_
>=
write_buf_
.
size
())
{
written_
=
0
;
write_buf_
.
clear
();
return
false
;
}
}
else
{
auto
err
=
get
<
sec
>
(
ret
);
CAF_LOG_DEBUG
(
"send failed"
<<
CAF_ARG
(
err
));
parent
.
application
().
handle_error
(
err
);
return
false
;
}
return
true
;
}
template
<
class
Parent
>
void
resolve
(
Parent
&
parent
,
const
std
::
string
&
path
,
actor
listener
)
{
parent
.
application
().
resolve
(
parent
,
path
,
listener
);
}
template
<
class
Parent
>
void
timeout
(
Parent
&
parent
,
atom_value
value
,
uint64_t
id
)
{
parent
.
application
().
timeout
(
*
this
,
value
,
id
);
}
template
<
class
Application
>
void
handle_error
(
Application
&
application
,
sec
code
)
{
application
.
handle_error
(
code
);
}
void
prepare_next_read
();
void
configure_read
(
net
::
receive_policy
::
config
cfg
);
void
write_packet
(
span
<
char
>
buf
);
private:
net
::
stream_socket
handle_
;
std
::
vector
<
char
>
read_buf_
;
std
::
vector
<
char
>
write_buf_
;
size_t
max_consecutive_reads_
;
size_t
read_threshold_
;
size_t
collected_
;
size_t
max_
;
net
::
receive_policy_flag
rd_flag_
;
size_t
written_
;
};
}
// namespace policy
}
// namespace caf
libcaf_net/src/scribe.cpp
0 → 100644
View file @
68451a12
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/policy/scribe.hpp"
#include <system_error>
#include "caf/config.hpp"
namespace
caf
{
namespace
policy
{
scribe
::
scribe
(
caf
::
net
::
stream_socket
handle
)
:
handle_
(
handle
),
max_consecutive_reads_
(
0
),
read_threshold_
(
1024
),
collected_
(
0
),
max_
(
1024
),
rd_flag_
(
net
::
receive_policy_flag
::
exactly
),
written_
(
0
)
{
// nop
}
void
scribe
::
prepare_next_read
()
{
read_buf_
.
clear
();
collected_
=
0
;
// This cast does nothing, but prevents a weird compiler error on GCC <= 4.9.
// TODO: remove cast when dropping support for GCC 4.9.
switch
(
static_cast
<
net
::
receive_policy_flag
>
(
rd_flag_
))
{
case
net
:
:
receive_policy_flag
::
exactly
:
if
(
read_buf_
.
size
()
!=
max_
)
read_buf_
.
resize
(
max_
);
read_threshold_
=
max_
;
break
;
case
net
:
:
receive_policy_flag
::
at_most
:
if
(
read_buf_
.
size
()
!=
max_
)
read_buf_
.
resize
(
max_
);
read_threshold_
=
1
;
break
;
case
net
:
:
receive_policy_flag
::
at_least
:
{
// read up to 10% more, but at least allow 100 bytes more
auto
max_size
=
max_
+
std
::
max
<
size_t
>
(
100
,
max_
/
10
);
if
(
read_buf_
.
size
()
!=
max_size
)
read_buf_
.
resize
(
max_size
);
read_threshold_
=
max_
;
break
;
}
}
}
void
scribe
::
configure_read
(
net
::
receive_policy
::
config
cfg
)
{
rd_flag_
=
cfg
.
first
;
max_
=
cfg
.
second
;
prepare_next_read
();
}
void
scribe
::
write_packet
(
span
<
char
>
buf
)
{
write_buf_
.
insert
(
write_buf_
.
end
(),
buf
.
begin
(),
buf
.
end
());
}
}
// namespace policy
}
// namespace caf
libcaf_net/test/scribe.cpp
0 → 100644
View file @
68451a12
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 scribe_policy
#include "caf/policy/scribe.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/make_actor.hpp"
#include "caf/net/actor_proxy_impl.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/stream_socket.hpp"
using
namespace
caf
;
using
namespace
caf
::
net
;
namespace
{
constexpr
string_view
hello_manager
=
"hello manager!"
;
struct
fixture
:
test_coordinator_fixture
<>
,
host_fixture
{
fixture
()
{
mpx
=
std
::
make_shared
<
multiplexer
>
();
if
(
auto
err
=
mpx
->
init
())
CAF_FAIL
(
"mpx->init failed: "
<<
sys
.
render
(
err
));
}
bool
handle_io_event
()
override
{
mpx
->
handle_updates
();
return
mpx
->
poll_once
(
false
);
}
multiplexer_ptr
mpx
;
};
class
dummy_application
{
public:
dummy_application
(
std
::
shared_ptr
<
std
::
vector
<
char
>>
rec_buf
)
:
rec_buf_
(
std
::
move
(
rec_buf
)){
// nop
};
~
dummy_application
()
=
default
;
template
<
class
Parent
>
error
init
(
Parent
&
)
{
return
none
;
}
template
<
class
Transport
>
void
write_message
(
Transport
&
transport
,
std
::
unique_ptr
<
endpoint_manager
::
message
>
msg
)
{
transport
.
write_packet
(
msg
->
payload
);
}
template
<
class
Parent
>
void
handle_data
(
Parent
&
,
span
<
char
>
data
)
{
rec_buf_
->
clear
();
rec_buf_
->
insert
(
rec_buf_
->
begin
(),
data
.
begin
(),
data
.
end
());
}
template
<
class
Manager
>
void
resolve
(
Manager
&
manager
,
const
std
::
string
&
path
,
actor
listener
)
{
actor_id
aid
=
42
;
auto
hid
=
"0011223344556677889900112233445566778899"
;
auto
nid
=
unbox
(
make_node_id
(
42
,
hid
));
actor_config
cfg
;
auto
p
=
make_actor
<
actor_proxy_impl
,
strong_actor_ptr
>
(
aid
,
nid
,
&
manager
.
system
(),
cfg
,
&
manager
);
anon_send
(
listener
,
resolve_atom
::
value
,
std
::
move
(
path
),
p
);
}
template
<
class
Transport
>
void
timeout
(
Transport
&
,
atom_value
,
uint64_t
)
{
// nop
}
void
handle_error
(
sec
)
{
// nop
}
static
expected
<
std
::
vector
<
char
>>
serialize
(
actor_system
&
sys
,
const
type_erased_tuple
&
x
)
{
std
::
vector
<
char
>
result
;
binary_serializer
sink
{
sys
,
result
};
if
(
auto
err
=
message
::
save
(
sink
,
x
))
return
err
;
return
result
;
}
private:
std
::
shared_ptr
<
std
::
vector
<
char
>>
rec_buf_
;
};
}
// namespace
CAF_TEST_FIXTURE_SCOPE
(
endpoint_manager_tests
,
fixture
)
CAF_TEST
(
receive
)
{
std
::
vector
<
char
>
read_buf
(
1024
);
CAF_CHECK_EQUAL
(
mpx
->
num_socket_managers
(),
1u
);
auto
buf
=
std
::
make_shared
<
std
::
vector
<
char
>>
();
auto
sockets
=
unbox
(
make_stream_socket_pair
());
nonblocking
(
sockets
.
second
,
true
);
CAF_CHECK_EQUAL
(
read
(
sockets
.
second
,
read_buf
.
data
(),
read_buf
.
size
()),
sec
::
unavailable_or_would_block
);
auto
guard
=
detail
::
make_scope_guard
([
&
]
{
close
(
sockets
.
second
);
});
CAF_MESSAGE
(
"configure scribe_policy"
);
policy
::
scribe
scribe
{
sockets
.
first
};
scribe
.
configure_read
(
net
::
receive_policy
::
exactly
(
hello_manager
.
size
()));
auto
mgr
=
make_endpoint_manager
(
mpx
,
sys
,
scribe
,
dummy_application
{
buf
});
CAF_CHECK_EQUAL
(
mgr
->
init
(),
none
);
mpx
->
handle_updates
();
CAF_CHECK_EQUAL
(
mpx
->
num_socket_managers
(),
2u
);
CAF_MESSAGE
(
"sending data to scribe_policy"
);
CAF_CHECK_EQUAL
(
write
(
sockets
.
second
,
hello_manager
.
data
(),
hello_manager
.
size
()),
hello_manager
.
size
());
run
();
CAF_CHECK_EQUAL
(
string_view
(
buf
->
data
(),
buf
->
size
()),
hello_manager
);
}
CAF_TEST
(
resolve
and
proxy
communication
)
{
std
::
vector
<
char
>
read_buf
(
1024
);
auto
buf
=
std
::
make_shared
<
std
::
vector
<
char
>>
();
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
,
policy
::
scribe
{
sockets
.
first
},
dummy_application
{
buf
});
CAF_CHECK_EQUAL
(
mgr
->
init
(),
none
);
mpx
->
handle_updates
();
run
();
mgr
->
resolve
(
"/id/42"
,
self
);
run
();
self
->
receive
(
[
&
](
resolve_atom
,
const
std
::
string
&
,
const
strong_actor_ptr
&
p
)
{
CAF_MESSAGE
(
"got a proxy, send a message to it"
);
self
->
send
(
actor_cast
<
actor
>
(
p
),
"hello proxy!"
);
},
after
(
std
::
chrono
::
seconds
(
0
))
>>
[
&
]
{
CAF_FAIL
(
"manager did not respond with a proxy."
);
});
run
();
auto
read_res
=
read
(
sockets
.
second
,
read_buf
.
data
(),
read_buf
.
size
());
if
(
!
holds_alternative
<
size_t
>
(
read_res
))
CAF_FAIL
(
"read() returned an error: "
<<
sys
.
render
(
get
<
sec
>
(
read_res
)));
read_buf
.
resize
(
get
<
size_t
>
(
read_res
));
CAF_MESSAGE
(
"receive buffer contains "
<<
read_buf
.
size
()
<<
" bytes"
);
message
msg
;
binary_deserializer
source
{
sys
,
read_buf
};
CAF_CHECK_EQUAL
(
source
(
msg
),
none
);
if
(
msg
.
match_elements
<
std
::
string
>
())
CAF_CHECK_EQUAL
(
msg
.
get_as
<
std
::
string
>
(
0
),
"hello proxy!"
);
else
CAF_ERROR
(
"expected a string, got: "
<<
to_string
(
msg
));
}
CAF_TEST_FIXTURE_SCOPE_END
()
libcaf_net/test/string_application.cpp
0 → 100644
View file @
68451a12
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 string_application
#include "caf/net/endpoint_manager.hpp"
#include <caf/policy/scribe.hpp>
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/make_actor.hpp"
#include "caf/net/actor_proxy_impl.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/stream_socket.hpp"
using
namespace
caf
;
using
namespace
caf
::
net
;
using
namespace
caf
::
policy
;
namespace
{
string_view
hello_manager
{
"hello manager!"
};
string_view
hello_test
{
"hello test!"
};
struct
fixture
:
test_coordinator_fixture
<>
,
host_fixture
{
fixture
()
{
mpx
=
std
::
make_shared
<
multiplexer
>
();
if
(
auto
err
=
mpx
->
init
())
CAF_FAIL
(
"mpx->init failed: "
<<
sys
.
render
(
err
));
}
bool
handle_io_event
()
override
{
mpx
->
handle_updates
();
return
mpx
->
poll_once
(
false
);
}
multiplexer_ptr
mpx
;
};
struct
string_application_header
{
uint32_t
payload
;
};
/// @relates header
template
<
class
Inspector
>
typename
Inspector
::
result_type
inspect
(
Inspector
&
f
,
string_application_header
&
hdr
)
{
return
f
(
meta
::
type_name
(
"sa_header"
),
hdr
.
payload
);
}
class
string_application
{
public:
using
header_type
=
string_application_header
;
string_application
(
actor_system
&
sys
,
std
::
shared_ptr
<
std
::
vector
<
char
>>
buf
)
:
sys_
(
sys
),
buf_
(
std
::
move
(
buf
))
{
// nop
}
template
<
class
Parent
>
error
init
(
Parent
&
)
{
return
none
;
}
template
<
class
Parent
>
void
handle_packet
(
Parent
&
,
header_type
&
,
span
<
char
>
payload
)
{
binary_deserializer
source
{
sys_
,
payload
.
data
(),
payload
.
size
()};
message
msg
;
if
(
auto
err
=
msg
.
load
(
source
))
CAF_FAIL
(
"unable to deserialize message: "
<<
err
);
if
(
!
msg
.
match_elements
<
std
::
string
>
())
CAF_FAIL
(
"unexpected message: "
<<
msg
);
auto
&
str
=
msg
.
get_as
<
std
::
string
>
(
0
);
buf_
->
insert
(
buf_
->
end
(),
str
.
begin
(),
str
.
end
());
}
template
<
class
Parent
>
void
write_message
(
Parent
&
parent
,
std
::
unique_ptr
<
net
::
endpoint_manager
::
message
>
msg
)
{
std
::
vector
<
char
>
buf
;
header_type
header
{
static_cast
<
uint32_t
>
(
msg
->
payload
.
size
())};
buf
.
resize
(
sizeof
(
header_type
));
memcpy
(
buf
.
data
(),
&
header
,
buf
.
size
());
buf
.
insert
(
buf
.
end
(),
msg
->
payload
.
begin
(),
msg
->
payload
.
end
());
parent
.
write_packet
(
buf
);
}
static
expected
<
std
::
vector
<
char
>>
serialize
(
actor_system
&
sys
,
const
type_erased_tuple
&
x
)
{
std
::
vector
<
char
>
result
;
binary_serializer
sink
{
sys
,
result
};
if
(
auto
err
=
message
::
save
(
sink
,
x
))
return
err
;
return
result
;
}
private:
actor_system
&
sys_
;
std
::
shared_ptr
<
std
::
vector
<
char
>>
buf_
;
};
class
stream_string_application
:
public
string_application
{
public:
stream_string_application
(
actor_system
&
sys
,
std
::
shared_ptr
<
std
::
vector
<
char
>>
buf
)
:
string_application
(
sys
,
std
::
move
(
buf
)),
await_payload_
(
false
)
{
// nop
}
template
<
class
Parent
>
error
init
(
Parent
&
parent
)
{
parent
.
transport
().
configure_read
(
net
::
receive_policy
::
exactly
(
sizeof
(
header_type
)));
return
string_application
::
init
(
parent
);
}
template
<
class
Parent
>
void
handle_data
(
Parent
&
parent
,
span
<
char
>
data
)
{
if
(
await_payload_
)
{
handle_packet
(
parent
,
header_
,
data
);
await_payload_
=
false
;
}
else
{
if
(
data
.
size
()
!=
sizeof
(
header_type
))
CAF_FAIL
(
""
);
memcpy
(
&
header_
,
data
.
data
(),
sizeof
(
header_type
));
if
(
header_
.
payload
==
0
)
handle_packet
(
parent
,
header_
,
span
<
char
>
{});
else
parent
.
configure_read
(
net
::
receive_policy
::
exactly
(
header_
.
payload
));
await_payload_
=
true
;
}
}
template
<
class
Manager
>
void
resolve
(
Manager
&
manager
,
const
std
::
string
&
path
,
actor
listener
)
{
actor_id
aid
=
42
;
auto
hid
=
"0011223344556677889900112233445566778899"
;
auto
nid
=
unbox
(
make_node_id
(
aid
,
hid
));
actor_config
cfg
;
auto
p
=
make_actor
<
net
::
actor_proxy_impl
,
strong_actor_ptr
>
(
aid
,
nid
,
&
manager
.
system
(),
cfg
,
&
manager
);
anon_send
(
listener
,
resolve_atom
::
value
,
std
::
move
(
path
),
p
);
}
template
<
class
Transport
>
void
timeout
(
Transport
&
,
atom_value
,
uint64_t
)
{
// nop
}
void
handle_error
(
sec
)
{
// nop
}
private:
header_type
header_
;
bool
await_payload_
;
};
}
// namespace
CAF_TEST_FIXTURE_SCOPE
(
endpoint_manager_tests
,
fixture
)
CAF_TEST
(
receive
)
{
std
::
vector
<
char
>
read_buf
(
1024
);
CAF_CHECK_EQUAL
(
mpx
->
num_socket_managers
(),
1u
);
auto
buf
=
std
::
make_shared
<
std
::
vector
<
char
>>
();
auto
sockets
=
unbox
(
make_stream_socket_pair
());
nonblocking
(
sockets
.
second
,
true
);
CAF_CHECK_EQUAL
(
read
(
sockets
.
second
,
read_buf
.
data
(),
read_buf
.
size
()),
sec
::
unavailable_or_would_block
);
CAF_MESSAGE
(
"adding both endpoint managers"
);
auto
mgr1
=
make_endpoint_manager
(
mpx
,
sys
,
policy
::
scribe
{
sockets
.
first
},
stream_string_application
{
sys
,
buf
});
CAF_CHECK_EQUAL
(
mgr1
->
init
(),
none
);
mpx
->
handle_updates
();
CAF_CHECK_EQUAL
(
mpx
->
num_socket_managers
(),
2u
);
auto
mgr2
=
make_endpoint_manager
(
mpx
,
sys
,
policy
::
scribe
{
sockets
.
second
},
stream_string_application
{
sys
,
buf
});
CAF_CHECK_EQUAL
(
mgr2
->
init
(),
none
);
mpx
->
handle_updates
();
CAF_CHECK_EQUAL
(
mpx
->
num_socket_managers
(),
3u
);
CAF_MESSAGE
(
"resolve actor-proxy"
);
mgr1
->
resolve
(
"/id/42"
,
self
);
run
();
self
->
receive
(
[
&
](
resolve_atom
,
const
std
::
string
&
,
const
strong_actor_ptr
&
p
)
{
CAF_MESSAGE
(
"got a proxy, send a message to it"
);
self
->
send
(
actor_cast
<
actor
>
(
p
),
"hello proxy!"
);
},
after
(
std
::
chrono
::
seconds
(
0
))
>>
[
&
]
{
CAF_FAIL
(
"manager did not respond with a proxy."
);
});
run
();
CAF_CHECK_EQUAL
(
string_view
(
buf
->
data
(),
buf
->
size
()),
"hello 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