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
8e2a5aa1
Commit
8e2a5aa1
authored
Aug 05, 2022
by
Dominik Charousset
Browse files
Options
Browse Files
Download
Plain Diff
Merge branch 'topic/neverlord/wss'
parents
888058a9
0744bcc7
Changes
6
Hide whitespace changes
Inline
Side-by-side
Showing
6 changed files
with
231 additions
and
47 deletions
+231
-47
examples/web_socket/hello-client.cpp
examples/web_socket/hello-client.cpp
+17
-27
libcaf_core/caf/uri.hpp
libcaf_core/caf/uri.hpp
+14
-0
libcaf_core/src/uri.cpp
libcaf_core/src/uri.cpp
+21
-0
libcaf_net/CMakeLists.txt
libcaf_net/CMakeLists.txt
+1
-0
libcaf_net/caf/net/web_socket/connect.hpp
libcaf_net/caf/net/web_socket/connect.hpp
+88
-20
libcaf_net/src/net/web_socket/connect.cpp
libcaf_net/src/net/web_socket/connect.cpp
+90
-0
No files found.
examples/web_socket/hello-client.cpp
View file @
8e2a5aa1
...
...
@@ -13,15 +13,11 @@
#include <iostream>
#include <utility>
static
constexpr
uint16_t
default_port
=
8080
;
struct
config
:
caf
::
actor_system_config
{
config
()
{
opt_group
{
custom_options_
,
"global"
}
//
.
add
<
std
::
string
>
(
"host"
,
"TCP endpoint to connect to"
)
.
add
<
uint16_t
>
(
"port,p"
,
"port for connecting to the host"
)
.
add
<
std
::
string
>
(
"endpoint"
,
"sets the Request-URI field"
)
.
add
<
std
::
string
>
(
"protocols"
,
"sets the Sec-WebSocket-Protocol field"
)
opt_group
{
custom_options_
,
"global"
}
.
add
<
caf
::
uri
>
(
"server,s"
,
"URI for connecting to the server"
)
.
add
<
std
::
string
>
(
"protocols,p"
,
"sets the Sec-WebSocket-Protocol field"
)
.
add
<
size_t
>
(
"max,m"
,
"maximum number of message to receive"
);
}
};
...
...
@@ -30,34 +26,27 @@ int caf_main(caf::actor_system& sys, const config& cfg) {
namespace
cn
=
caf
::
net
;
namespace
ws
=
cn
::
web_socket
;
// Sanity checking.
auto
host
=
caf
::
get_or
(
cfg
,
"host"
,
""
);
if
(
host
.
empty
())
{
std
::
cerr
<<
"*** m
issing mandatory argument: host
\n
"
;
auto
server
=
caf
::
get_or
(
cfg
,
"server"
,
caf
::
uri
{}
);
if
(
!
server
.
valid
())
{
std
::
cerr
<<
"*** m
andatory argument server missing or invalid
\n
"
;
return
EXIT_FAILURE
;
}
// Ask user for the hello message.
std
::
string
hello
;
std
::
cout
<<
"Please enter a hello message for the server: "
<<
std
::
flush
;
std
::
getline
(
std
::
cin
,
hello
);
//
Open up the TCP connection
.
auto
port
=
caf
::
get_or
(
cfg
,
"port"
,
default_port
);
cn
::
tcp_stream_socket
fd
;
if
(
auto
maybe_fd
=
cn
::
make_connected_tcp_stream_socket
(
host
,
port
))
{
std
::
cout
<<
"*** connected to "
<<
host
<<
':'
<<
port
<<
'\n'
;
fd
=
std
::
move
(
*
maybe_fd
)
;
}
else
{
std
::
cerr
<<
"*** unable to connect to "
<<
host
<<
':'
<<
port
<<
": "
<<
to_string
(
maybe_fd
.
error
())
<<
'\n'
;
//
Spin up the WebSocket
.
auto
hs_setup
=
[
&
cfg
](
ws
::
handshake
&
hs
)
{
if
(
auto
str
=
caf
::
get_as
<
std
::
string
>
(
cfg
,
"protocols"
)
;
str
&&
!
str
->
empty
())
hs
.
protocols
(
std
::
move
(
*
str
))
;
}
;
auto
conn
=
ws
::
connect
(
sys
,
server
,
hs_setup
);
if
(
!
conn
)
{
std
::
cerr
<<
"*** failed to connect: "
<<
to_string
(
conn
.
error
())
<<
'\n'
;
return
EXIT_FAILURE
;
}
// Spin up the WebSocket.
ws
::
handshake
hs
;
hs
.
host
(
host
);
hs
.
endpoint
(
caf
::
get_or
(
cfg
,
"endpoint"
,
"/"
));
if
(
auto
str
=
caf
::
get_as
<
std
::
string
>
(
cfg
,
"protocols"
);
str
&&
!
str
->
empty
())
hs
.
protocols
(
std
::
move
(
*
str
));
ws
::
connect
(
sys
,
fd
,
std
::
move
(
hs
),
[
&
](
const
ws
::
connect_event_t
&
conn
)
{
conn
->
run
([
&
](
const
ws
::
connect_event_t
&
conn
)
{
sys
.
spawn
([
conn
,
hello
](
caf
::
event_based_actor
*
self
)
{
auto
[
pull
,
push
]
=
conn
.
data
();
// Print everything from the server to stdout.
...
...
@@ -73,6 +62,7 @@ int caf_main(caf::actor_system& sys, const config& cfg) {
return
std
::
move
(
in
).
as_observable
();
}
})
.
do_finally
([]
{
std
::
cout
<<
"Server has closed the connection
\n
"
;
})
.
for_each
([](
const
ws
::
frame
&
msg
)
{
if
(
msg
.
is_text
())
{
std
::
cout
<<
"Server: "
<<
msg
.
as_text
()
<<
'\n'
;
...
...
libcaf_core/caf/uri.hpp
View file @
8e2a5aa1
...
...
@@ -91,6 +91,9 @@ public:
/// The fragment component.
std
::
string
fragment
;
/// Offset to the path in `str`.
size_t
path_offset
=
0
;
// -- properties -----------------------------------------------------------
bool
valid
()
const
noexcept
{
...
...
@@ -101,6 +104,10 @@ public:
return
rc_
.
load
()
==
1
;
}
std
::
string_view
str_after_path_offset
()
const
noexcept
{
return
{
str
.
c_str
()
+
path_offset
,
str
.
size
()
-
path_offset
};
}
// -- modifiers ------------------------------------------------------------
/// Assembles the human-readable string representation for this URI.
...
...
@@ -182,6 +189,13 @@ public:
return
impl_
->
fragment
;
}
/// Returns the host sub-component of the authority as string.
std
::
string
host_str
()
const
;
/// Returns the path, query and fragment components (as they appear in the
/// encoded URI) with a leading '/'.
std
::
string
path_query_fragment
()
const
;
/// Returns a hash code over all components.
size_t
hash_code
()
const
noexcept
;
...
...
libcaf_core/src/uri.cpp
View file @
8e2a5aa1
...
...
@@ -37,10 +37,12 @@ void uri::impl_type::assemble_str() {
str
+=
':'
;
if
(
authority
.
empty
())
{
CAF_ASSERT
(
!
path
.
empty
());
path_offset
=
str
.
size
();
uri
::
encode
(
str
,
path
,
true
);
}
else
{
str
+=
"//"
;
str
+=
to_string
(
authority
);
path_offset
=
str
.
size
();
if
(
!
path
.
empty
())
{
str
+=
'/'
;
uri
::
encode
(
str
,
path
,
true
);
...
...
@@ -74,6 +76,25 @@ uri::uri(impl_ptr ptr) : impl_(std::move(ptr)) {
CAF_ASSERT
(
impl_
!=
nullptr
);
}
std
::
string
uri
::
host_str
()
const
{
const
auto
&
host
=
impl_
->
authority
.
host
;
if
(
std
::
holds_alternative
<
std
::
string
>
(
host
))
{
return
std
::
get
<
std
::
string
>
(
host
);
}
else
{
return
to_string
(
std
::
get
<
ip_address
>
(
host
));
}
}
std
::
string
uri
::
path_query_fragment
()
const
{
std
::
string
result
;
auto
sub_str
=
impl_
->
str_after_path_offset
();
if
(
sub_str
.
empty
()
||
sub_str
[
0
]
!=
'/'
)
{
result
+=
'/'
;
}
result
.
insert
(
result
.
begin
(),
sub_str
.
begin
(),
sub_str
.
end
());
return
result
;
}
size_t
uri
::
hash_code
()
const
noexcept
{
return
hash
::
fnv
<
size_t
>::
compute
(
str
());
}
...
...
libcaf_net/CMakeLists.txt
View file @
8e2a5aa1
...
...
@@ -78,6 +78,7 @@ caf_add_component(
src/net/this_host.cpp
src/net/udp_datagram_socket.cpp
src/net/web_socket/client.cpp
src/net/web_socket/connect.cpp
src/net/web_socket/default_trait.cpp
src/net/web_socket/frame.cpp
src/net/web_socket/framing.cpp
...
...
libcaf_net/caf/net/web_socket/connect.hpp
View file @
8e2a5aa1
...
...
@@ -4,12 +4,19 @@
#pragma once
#include "caf/async/spsc_buffer.hpp"
#include "caf/callback.hpp"
#include "caf/cow_tuple.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/disposable.hpp"
#include "caf/expected.hpp"
#include "caf/net/ssl/connection.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/web_socket/client.hpp"
#include "caf/net/web_socket/default_trait.hpp"
#include "caf/net/web_socket/flow_bridge.hpp"
#include "caf/net/web_socket/frame.hpp"
#include "caf/net/web_socket/handshake.hpp"
#include "caf/uri.hpp"
#include <utility>
namespace
caf
::
net
::
web_socket
{
...
...
@@ -18,27 +25,88 @@ using connect_event_t
=
cow_tuple
<
async
::
consumer_resource
<
frame
>
,
// Socket to App.
async
::
producer_resource
<
frame
>>
;
// App to Socket.
/// Tries to establish a WebSocket connection on @p fd.
/// @param sys The host system.
/// Wraps a successful connection setup.
class
connect_state
{
public:
connect_state
()
=
delete
;
connect_state
(
connect_state
&&
)
=
default
;
connect_state
(
const
connect_state
&
)
=
default
;
connect_state
&
operator
=
(
connect_state
&&
)
=
default
;
connect_state
&
operator
=
(
const
connect_state
&
)
=
default
;
connect_state
(
disposable
worker
,
connect_event_t
event
)
:
worker_
(
worker
),
event_
(
std
::
move
(
event
))
{
// nop
}
template
<
class
Init
>
disposable
run
(
Init
init
)
{
init
(
std
::
move
(
event_
));
return
worker_
;
}
private:
disposable
worker_
;
connect_event_t
event_
;
};
}
// namespace caf::net::web_socket
namespace
caf
::
detail
{
using
ws_handshake_setup
=
callback
<
void
(
net
::
web_socket
::
handshake
&
)
>
;
net
::
web_socket
::
connect_state
CAF_NET_EXPORT
ws_do_connect
(
actor_system
&
sys
,
net
::
stream_socket
fd
,
net
::
web_socket
::
handshake
&
hs
);
net
::
web_socket
::
connect_state
CAF_NET_EXPORT
ws_do_connect
(
actor_system
&
sys
,
net
::
ssl
::
connection
conn
,
net
::
web_socket
::
handshake
&
hs
);
expected
<
net
::
web_socket
::
connect_state
>
CAF_NET_EXPORT
ws_connect_impl
(
actor_system
&
sys
,
const
caf
::
uri
&
dst
,
ws_handshake_setup
&
setup
);
}
// namespace caf::detail
namespace
caf
::
net
::
web_socket
{
/// Starts a WebSocket connection on @p fd.
/// @param sys The parent system.
/// @param fd An connected socket.
/// @param init Function object for setting up the created flows.
template
<
class
Transport
=
stream_transport
,
class
Socket
,
class
Init
>
void
connect
(
actor_system
&
sys
,
Socket
fd
,
handshake
hs
,
Init
init
)
{
// TODO: connect() should return a disposable to stop the WebSocket.
using
trait_t
=
default_trait
;
template
<
class
Init
>
disposable
connect
(
actor_system
&
sys
,
stream_socket
fd
,
handshake
hs
,
Init
init
)
{
static_assert
(
std
::
is_invocable_v
<
Init
,
connect_event_t
&&>
,
"invalid signature found for init"
);
auto
[
ws_pull
,
app_push
]
=
async
::
make_spsc_buffer_resource
<
frame
>
();
auto
[
app_pull
,
ws_push
]
=
async
::
make_spsc_buffer_resource
<
frame
>
();
using
conn_t
=
flow_connector_trivial_impl
<
trait_t
>
;
auto
mpx
=
sys
.
network_manager
().
mpx_ptr
();
auto
conn
=
std
::
make_shared
<
conn_t
>
(
std
::
move
(
ws_pull
),
std
::
move
(
ws_push
));
auto
bridge
=
flow_bridge
<
trait_t
>::
make
(
mpx
,
std
::
move
(
conn
));
auto
impl
=
client
::
make
(
std
::
move
(
hs
),
std
::
move
(
bridge
));
auto
transport
=
Transport
::
make
(
fd
,
std
::
move
(
impl
));
auto
ptr
=
socket_manager
::
make
(
mpx
,
std
::
move
(
transport
));
mpx
->
start
(
ptr
);
init
(
connect_event_t
{
app_pull
,
app_push
});
return
detail
::
ws_do_connect
(
sys
,
fd
,
hs
).
run
(
std
::
move
(
init
));
}
/// Starts a WebSocket connection on @p fd.
/// @param sys The parent system.
/// @param conn An established TCP with an TLS connection attached to it.
/// @param init Function object for setting up the created flows.
template
<
class
Init
>
disposable
connect
(
actor_system
&
sys
,
ssl
::
connection
conn
,
handshake
hs
,
Init
init
)
{
static_assert
(
std
::
is_invocable_v
<
Init
,
connect_event_t
&&>
,
"invalid signature found for init"
);
return
detail
::
ws_do_connect
(
sys
,
std
::
move
(
conn
),
hs
).
run
(
std
::
move
(
init
));
}
/// Tries to connect to the host from the URI.
/// @param sys The parent system.
/// @param dst Encodes the destination host. URI scheme must be `ws` or `wss`.
/// @note Blocks the caller while trying to establish a TCP connection.
expected
<
connect_state
>
connect
(
actor_system
&
sys
,
const
caf
::
uri
&
dst
);
template
<
class
HandshakeSetup
>
expected
<
connect_state
>
connect
(
actor_system
&
sys
,
const
caf
::
uri
&
dst
,
HandshakeSetup
&&
setup
)
{
static_assert
(
std
::
is_invocable_v
<
HandshakeSetup
,
handshake
&>
,
"invalid signature found for the handshake setup function"
);
auto
cb
=
make_callback
(
std
::
forward
<
HandshakeSetup
>
(
setup
));
return
detail
::
ws_connect_impl
(
sys
,
dst
,
cb
);
}
}
// namespace caf::net::web_socket
libcaf_net/src/net/web_socket/connect.cpp
0 → 100644
View file @
8e2a5aa1
// 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.
#include "caf/net/web_socket/connect.hpp"
#include "caf/net/flow_connector.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/ssl/context.hpp"
#include "caf/net/ssl/transport.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/net/web_socket/client.hpp"
#include "caf/net/web_socket/default_trait.hpp"
#include "caf/net/web_socket/flow_bridge.hpp"
namespace
ws
=
caf
::
net
::
web_socket
;
namespace
caf
::
detail
{
template
<
class
Transport
,
class
Connection
>
ws
::
connect_state
ws_do_connect_impl
(
actor_system
&
sys
,
Connection
conn
,
ws
::
handshake
&
hs
)
{
using
trait_t
=
ws
::
default_trait
;
using
connector_t
=
net
::
flow_connector_trivial_impl
<
trait_t
>
;
auto
[
ws_pull
,
app_push
]
=
async
::
make_spsc_buffer_resource
<
ws
::
frame
>
();
auto
[
app_pull
,
ws_push
]
=
async
::
make_spsc_buffer_resource
<
ws
::
frame
>
();
auto
mpx
=
sys
.
network_manager
().
mpx_ptr
();
auto
connector
=
std
::
make_shared
<
connector_t
>
(
std
::
move
(
ws_pull
),
std
::
move
(
ws_push
));
auto
bridge
=
ws
::
flow_bridge
<
trait_t
>::
make
(
mpx
,
std
::
move
(
connector
));
auto
impl
=
ws
::
client
::
make
(
std
::
move
(
hs
),
std
::
move
(
bridge
));
auto
fd
=
conn
.
fd
();
auto
transport
=
Transport
::
make
(
std
::
move
(
conn
),
std
::
move
(
impl
));
transport
->
active_policy
().
connect
(
fd
);
auto
ptr
=
net
::
socket_manager
::
make
(
mpx
,
std
::
move
(
transport
));
mpx
->
start
(
ptr
);
return
ws
::
connect_state
{
ptr
->
as_disposable
(),
ws
::
connect_event_t
{
app_pull
,
app_push
}};
}
ws
::
connect_state
ws_do_connect
(
actor_system
&
sys
,
net
::
stream_socket
fd
,
ws
::
handshake
&
hs
)
{
return
ws_do_connect_impl
<
net
::
stream_transport
>
(
sys
,
fd
,
hs
);
}
ws
::
connect_state
ws_do_connect
(
actor_system
&
sys
,
net
::
ssl
::
connection
conn
,
ws
::
handshake
&
hs
)
{
return
ws_do_connect_impl
<
net
::
ssl
::
transport
>
(
sys
,
std
::
move
(
conn
),
hs
);
}
expected
<
ws
::
connect_state
>
ws_connect_impl
(
actor_system
&
sys
,
const
caf
::
uri
&
dst
,
ws_handshake_setup
&
setup
)
{
if
(
dst
.
scheme
()
!=
"ws"
&&
dst
.
scheme
()
!=
"wss"
)
{
return
{
make_error
(
sec
::
invalid_argument
,
"URI must use ws or wss scheme"
)};
}
auto
fd
=
net
::
make_connected_tcp_stream_socket
(
dst
.
authority
());
if
(
!
fd
)
{
return
{
std
::
move
(
fd
.
error
())};
}
net
::
web_socket
::
handshake
hs
;
hs
.
host
(
dst
.
host_str
());
hs
.
endpoint
(
dst
.
path_query_fragment
());
setup
(
hs
);
if
(
dst
.
scheme
()
==
"ws"
)
{
return
{
ws_do_connect
(
sys
,
*
fd
,
hs
)};
}
else
{
auto
ctx
=
net
::
ssl
::
context
::
make_client
(
net
::
ssl
::
tls
::
any
);
if
(
!
ctx
)
{
return
{
std
::
move
(
ctx
.
error
())};
}
auto
conn
=
ctx
->
new_connection
(
*
fd
);
if
(
!
conn
)
{
return
{
std
::
move
(
conn
.
error
())};
}
return
{
ws_do_connect
(
sys
,
std
::
move
(
*
conn
),
hs
)};
}
}
}
// namespace caf::detail
namespace
caf
::
net
::
web_socket
{
expected
<
connect_state
>
connect
(
actor_system
&
sys
,
const
caf
::
uri
&
dst
)
{
auto
cb
=
make_callback
([](
net
::
web_socket
::
handshake
&
)
{});
return
detail
::
ws_connect_impl
(
sys
,
dst
,
cb
);
}
}
// namespace caf::net::web_socket
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