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
4d03a100
Commit
4d03a100
authored
Mar 13, 2021
by
Dominik Charousset
Browse files
Options
Browse Files
Download
Plain Diff
Merge branch 'topic/neverlord/websocket-client'
parents
f334ec4b
d0b7e4bf
Changes
18
Expand all
Show whitespace changes
Inline
Side-by-side
Showing
18 changed files
with
1812 additions
and
654 deletions
+1812
-654
examples/CMakeLists.txt
examples/CMakeLists.txt
+1
-0
examples/net/web-socket-calculator.cpp
examples/net/web-socket-calculator.cpp
+116
-71
examples/net/web-socket-client.cpp
examples/net/web-socket-client.cpp
+296
-0
examples/net/web-socket-feed.cpp
examples/net/web-socket-feed.cpp
+4
-4
libcaf_net/CMakeLists.txt
libcaf_net/CMakeLists.txt
+4
-1
libcaf_net/caf/net/basp/header.hpp
libcaf_net/caf/net/basp/header.hpp
+0
-2
libcaf_net/caf/net/middleman.hpp
libcaf_net/caf/net/middleman.hpp
+8
-0
libcaf_net/caf/net/web_socket/client.hpp
libcaf_net/caf/net/web_socket/client.hpp
+176
-0
libcaf_net/caf/net/web_socket/framing.hpp
libcaf_net/caf/net/web_socket/framing.hpp
+319
-0
libcaf_net/caf/net/web_socket/handshake.hpp
libcaf_net/caf/net/web_socket/handshake.hpp
+154
-0
libcaf_net/caf/net/web_socket/server.hpp
libcaf_net/caf/net/web_socket/server.hpp
+237
-0
libcaf_net/caf/net/web_socket_framing.hpp
libcaf_net/caf/net/web_socket_framing.hpp
+4
-308
libcaf_net/caf/net/web_socket_server.hpp
libcaf_net/caf/net/web_socket_server.hpp
+4
-263
libcaf_net/src/net/web_socket/handshake.cpp
libcaf_net/src/net/web_socket/handshake.cpp
+262
-0
libcaf_net/test/net-test.hpp
libcaf_net/test/net-test.hpp
+9
-1
libcaf_net/test/net/web_socket/client.cpp
libcaf_net/test/net/web_socket/client.cpp
+121
-0
libcaf_net/test/net/web_socket/handshake.cpp
libcaf_net/test/net/web_socket/handshake.cpp
+93
-0
libcaf_net/test/net/web_socket/server.cpp
libcaf_net/test/net/web_socket/server.cpp
+4
-4
No files found.
examples/CMakeLists.txt
View file @
4d03a100
...
...
@@ -16,4 +16,5 @@ if(TARGET CAF::net)
endfunction
()
add_net_example
(
web-socket-calculator
)
add_net_example
(
web-socket-feed
)
add_net_example
(
web-socket-client
)
endif
()
examples/net/web-socket-calculator.cpp
View file @
4d03a100
...
...
@@ -18,8 +18,8 @@
// import asyncio
// import websockets
//
// line1 = '{
values = [ { "@type" = "task::addition", x = 17, y = 8 } ]
}\n'
// line2 = '{
values = [ { "@type" = "task::subtraction", x = 17, y = 8 } ]
}\n'
// line1 = '{
"@type": "addition", "x": 17, "y": 8
}\n'
// line2 = '{
"@type": "subtraction", "x": 17, "y": 8
}\n'
//
// async def hello():
// uri = "ws://localhost:8080"
...
...
@@ -39,63 +39,84 @@
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/byte_span.hpp"
#include "caf/caf_main.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/exec_main.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/json_reader.hpp"
#include "caf/json_writer.hpp"
#include "caf/mtl.hpp"
#include "caf/net/actor_shell.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/web_socket
_
server.hpp"
#include "caf/net/web_socket
/
server.hpp"
#include "caf/tag/mixed_message_oriented.hpp"
#include "caf/typed_actor.hpp"
#include "caf/typed_event_based_actor.hpp"
#include <cstdint>
// -- custom
message types ----------
-------------------------------------------
// -- custom
actor and message types
-------------------------------------------
// Usually, we prefer atoms to prefix certain operations. However, using custom
// message types provides a nicer interface for the text-based WebSocket
// communication.
// Internal interface to the worker.
using
calculator_actor
=
caf
::
typed_actor
<
caf
::
result
<
int32_t
>
(
caf
::
add_atom
,
int32_t
,
int32_t
),
caf
::
result
<
int32_t
>
(
caf
::
sub_atom
,
int32_t
,
int32_t
)
>
;
namespace
task
{
// Utility for assigning JSON "type names" to CAF atoms.
template
<
class
Atom
>
struct
json_name
;
struct
addition
{
int32_t
x
;
int32_t
y
;
template
<
>
struct
json_name
<
caf
::
add_atom
>
{
static
constexpr
caf
::
string_view
value
=
"addition"
;
};
template
<
class
Inspector
>
bool
inspect
(
Inspector
&
f
,
addition
&
x
)
{
return
f
.
object
(
x
).
fields
(
f
.
field
(
"x"
,
x
.
x
),
f
.
field
(
"y"
,
x
.
y
));
}
struct
subtraction
{
int32_t
x
;
int32_t
y
;
template
<
>
struct
json_name
<
caf
::
sub_atom
>
{
static
constexpr
caf
::
string_view
value
=
"subtraction"
;
};
template
<
class
Inspector
>
bool
inspect
(
Inspector
&
f
,
subtraction
&
x
)
{
return
f
.
object
(
x
).
fields
(
f
.
field
(
"x"
,
x
.
x
),
f
.
field
(
"y"
,
x
.
y
));
}
}
// namespace task
CAF_BEGIN_TYPE_ID_BLOCK
(
web_socket_calculator
,
caf
::
first_custom_type_id
)
CAF_ADD_TYPE_ID
(
web_socket_calculator
,
(
task
::
addition
))
CAF_ADD_TYPE_ID
(
web_socket_calculator
,
(
task
::
subtraction
))
// Internally, we use atom-prefixed message. Externally, we exchange JSON
// objects over a WebSocket. This adapter translates between an external client
// and a `calculator_actor`.
struct
adapter
{
// The `calculator_actor` always receives three arguments: one atom (add_atom
// or sub_atom) and two integers (x and y). When reading from the inspector,
// we require an object with a matching type name that has two fields: x and
// y. When used with a `json_reader`, the format we accept from clients is:
// `{"@type": <addition-or-subtraction>, "x": <int32>, "y": <int32>}`.
template
<
class
Inspector
,
class
Atom
>
bool
read
(
Inspector
&
f
,
Atom
,
int32_t
&
x
,
int32_t
&
y
)
const
{
auto
type_annotation
=
json_name
<
Atom
>::
value
;
return
f
.
assert_next_object_name
(
type_annotation
)
&&
f
.
virtual_object
(
type_annotation
)
.
fields
(
f
.
field
(
"x"
,
x
),
f
.
field
(
"y"
,
y
));
}
CAF_END_TYPE_ID_BLOCK
(
web_socket_calculator
)
// The `calculator_actor` always returns an integer result. We simply apply
// this to the inspector. For a `json_writer`, this simply converts the
// integer to a string.
template
<
class
Inspector
>
bool
write
(
Inspector
&
f
,
int32_t
result
)
const
{
return
f
.
apply
(
result
);
}
};
// -- implementation of the calculator actor -----------------------------------
caf
::
behavior
calculator
()
{
struct
calculator_state
{
static
inline
const
char
*
name
=
"calculator"
;
calculator_actor
::
behavior_type
make_behavior
()
{
return
{
[](
task
::
addition
input
)
{
return
input
.
x
+
input
.
y
;
},
[](
task
::
subtraction
input
)
{
return
input
.
x
-
input
.
y
;
},
[](
caf
::
add_atom
,
int32_t
x
,
int32_t
y
)
{
return
x
+
y
;
},
[](
caf
::
sub_atom
,
int32_t
x
,
int32_t
y
)
{
return
x
-
y
;
},
};
}
}
};
using
calculator_impl
=
calculator_actor
::
stateful_impl
<
calculator_state
>
;
// -- implementation of the WebSocket application ------------------------------
...
...
@@ -103,7 +124,7 @@ class app {
public:
// -- member types -----------------------------------------------------------
//
We expect a stream-oriented interface to the lower communication layer
s.
//
Tells CAF we expect a transport with text and binary message
s.
using
input_tag
=
caf
::
tag
::
mixed_message_oriented
;
// -- constants --------------------------------------------------------------
...
...
@@ -113,7 +134,7 @@ public:
// -- constructors, destructors, and assignment operators --------------------
explicit
app
(
ca
f
::
actor
worker
)
:
worker_
(
std
::
move
(
worker
))
{
explicit
app
(
ca
lculator_
actor
worker
)
:
worker_
(
std
::
move
(
worker
))
{
// nop
}
...
...
@@ -160,6 +181,15 @@ public:
<<
to_string
(
reason
)
<<
"
\n
"
;
}
template
<
class
LowerLayerPtr
>
static
void
send_error
(
LowerLayerPtr
down
,
const
caf
::
error
&
err
)
{
auto
str
=
to_string
(
err
);
down
->
begin_text_message
();
auto
&
buf
=
down
->
text_message_buffer
();
buf
.
insert
(
buf
.
end
(),
str
.
begin
(),
str
.
end
());
down
->
end_text_message
();
}
template
<
class
LowerLayerPtr
>
ptrdiff_t
consume_text
(
LowerLayerPtr
down
,
caf
::
string_view
text
)
{
// The other functions in this class provide mostly boilerplate code. Here
...
...
@@ -185,36 +215,45 @@ public:
caf
::
string_view
line
{
buf_
.
data
(),
static_cast
<
size_t
>
(
num_bytes
)
-
1
};
std
::
cout
<<
"*** [socket "
<<
down
->
handle
().
id
<<
"] INPUT: "
<<
line
<<
"
\n
"
;
caf
::
config_value
val
;
if
(
auto
parsed_val
=
caf
::
config_value
::
parse
(
line
))
{
val
=
std
::
move
(
*
parsed_val
);
}
else
{
down
->
abort_reason
(
std
::
move
(
parsed_val
.
error
()));
return
-
1
;
}
caf
::
config_value_reader
reader
{
&
val
};
caf
::
message
msg
;
if
(
!
reader
.
apply
(
msg
))
{
down
->
abort_reason
(
reader
.
get_error
());
return
-
1
;
}
// Dispatch message to worker.
self_
->
request
(
worker_
,
std
::
chrono
::
seconds
{
1
},
std
::
move
(
msg
))
.
then
(
[
this
,
down
](
int32_t
value
)
{
if
(
reader
.
load
(
line
))
{
auto
on_result
=
[
this
,
down
](
auto
&
...
xs
)
{
// Simply respond with the value as string, wrapped into a WebSocket
// text message frame.
auto
str_response
=
std
::
to_string
(
value
);
writer
.
reset
();
if
(
!
adapter
{}.
write
(
writer
,
xs
...))
{
std
::
cerr
<<
"*** [socket "
<<
down
->
handle
().
id
<<
"] failed to generate JSON response: "
<<
to_string
(
writer
.
get_error
())
<<
"
\n
"
;
down
->
abort_reason
(
caf
::
sec
::
runtime_error
);
return
;
}
auto
str_response
=
writer
.
str
();
std
::
cout
<<
"*** [socket "
<<
down
->
handle
().
id
<<
"] OUTPUT: "
<<
str_response
<<
"
\n
"
;
str_response
+=
'\n'
;
down
->
begin_text_message
();
auto
&
buf
=
down
->
text_message_buffer
();
buf
.
insert
(
buf
.
end
(),
str_response
.
begin
(),
str_response
.
end
());
down
->
end_text_message
();
},
[
down
](
caf
::
error
&
err
)
{
down
->
abort_reason
(
std
::
move
(
err
));
});
// Erase consumed data from the buffer.
};
auto
on_error
=
[
down
](
caf
::
error
&
err
)
{
send_error
(
down
,
err
);
down
->
abort_reason
(
std
::
move
(
err
));
};
auto
mtl
=
caf
::
make_mtl
(
self_
.
get
(),
adapter
{},
&
reader
);
if
(
!
mtl
.
try_request
(
worker_
,
std
::
chrono
::
seconds
(
1
),
on_result
,
on_error
))
{
std
::
cerr
<<
"*** [socket "
<<
down
->
handle
().
id
<<
"] unable to deserialize a message from the received "
"JSON line
\n
"
;
auto
reason
=
make_error
(
caf
::
sec
::
runtime_error
,
"found no matching handler"
);
send_error
(
down
,
reason
);
}
}
else
{
std
::
cerr
<<
"*** [socket "
<<
down
->
handle
().
id
<<
"] unable to parse received JSON line
\n
"
;
send_error
(
down
,
reader
.
get_error
());
}
buf_
.
erase
(
buf_
.
begin
(),
i
+
1
);
}
return
static_cast
<
ptrdiff_t
>
(
text
.
size
());
...
...
@@ -234,10 +273,16 @@ private:
std
::
vector
<
char
>
buf_
;
// Stores a handle to our worker.
ca
f
::
actor
worker_
;
ca
lculator_
actor
worker_
;
// Enables the application to send and receive actor messages.
caf
::
net
::
actor_shell_ptr
self_
;
// Enables us to read JSON input from a WebSocket connection.
caf
::
json_reader
reader
;
// Enables us to write JSON output to the WebSocket connection.
caf
::
json_writer
writer
;
};
// -- main ---------------------------------------------------------------------
...
...
@@ -267,13 +312,13 @@ int caf_main(caf::actor_system& sys, const config& cfg) {
return
EXIT_FAILURE
;
}
// Spawn our worker actor and initiate the protocol stack.
auto
worker
=
sys
.
spawn
(
calculator
);
auto
worker
=
sys
.
spawn
<
calculator_impl
>
(
);
auto
add_conn
=
[
worker
](
tcp_stream_socket
sock
,
multiplexer
*
mpx
)
{
return
make_socket_manager
<
app
,
web_socket
_
server
,
stream_transport
>
(
return
make_socket_manager
<
app
,
web_socket
::
server
,
stream_transport
>
(
sock
,
mpx
,
worker
);
};
sys
.
network_manager
().
make_acceptor
(
sock
,
add_conn
);
return
EXIT_SUCCESS
;
}
CAF_MAIN
(
caf
::
id_block
::
web_socket_calculator
,
caf
::
net
::
middleman
)
CAF_MAIN
(
caf
::
net
::
middleman
)
examples/net/web-socket-client.cpp
0 → 100644
View file @
4d03a100
// This example application connects to a WebSocket server and allows users to
// send arbitrary text.
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/byte_span.hpp"
#include "caf/caf_main.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/net/actor_shell.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/web_socket/client.hpp"
#include "caf/scoped_actor.hpp"
#include "caf/stateful_actor.hpp"
#include "caf/tag/mixed_message_oriented.hpp"
#include "caf/type_id.hpp"
#include "caf/typed_event_based_actor.hpp"
#include "caf/uri.hpp"
#include <cstdio>
#include <iostream>
#include <termios.h>
#include <unistd.h>
// -- custom actor and message types -------------------------------------------
CAF_BEGIN_TYPE_ID_BLOCK
(
web_socket_client
,
caf
::
first_custom_type_id
)
CAF_ADD_ATOM
(
web_socket_client
,
atom
,
attach
)
CAF_ADD_ATOM
(
web_socket_client
,
atom
,
binary
)
CAF_ADD_ATOM
(
web_socket_client
,
atom
,
detach
)
CAF_ADD_ATOM
(
web_socket_client
,
atom
,
input
)
CAF_ADD_ATOM
(
web_socket_client
,
atom
,
status
)
CAF_ADD_ATOM
(
web_socket_client
,
atom
,
text
)
CAF_END_TYPE_ID_BLOCK
(
web_socket_client
)
using
terminal_actor
=
caf
::
typed_actor
<
// Attaches the terminal to an actor. After receiving this message, the
// terminal actor forwards all user input to the attached input.
caf
::
result
<
void
>
(
atom
::
attach
,
caf
::
actor
),
// Releases the terminal again if the sender is the currently attached actor.
caf
::
result
<
void
>
(
atom
::
detach
),
// Displays binary data received from the server.
caf
::
result
<
void
>
(
atom
::
binary
,
caf
::
byte_buffer
),
// Displays a status message from the attached actor.
caf
::
result
<
void
>
(
atom
::
status
,
std
::
string
),
// Displays text data received from the server.
caf
::
result
<
void
>
(
atom
::
text
,
std
::
string
),
// Handles input from the command line.
caf
::
result
<
void
>
(
atom
::
input
,
std
::
string
)
>
;
// -- implementation of our WebSocket application ------------------------------
class
app
{
public:
// -- member types -----------------------------------------------------------
// Tells CAF we expect a transport with text and binary messages.
using
input_tag
=
caf
::
tag
::
mixed_message_oriented
;
// -- constructors, destructors, and assignment operators --------------------
app
(
terminal_actor
sink
)
:
sink_
(
sink
)
{
// nop
}
template
<
class
LowerLayerPtr
>
caf
::
error
init
(
caf
::
net
::
socket_manager
*
mgr
,
LowerLayerPtr
down
,
const
caf
::
settings
&
)
{
self_
=
mgr
->
make_actor_shell
(
down
);
self_
->
set_behavior
([
down
](
const
std
::
string
&
line
)
{
down
->
begin_text_message
();
auto
&
buf
=
down
->
text_message_buffer
();
buf
.
insert
(
buf
.
end
(),
line
.
begin
(),
line
.
end
());
down
->
end_text_message
();
});
self_
->
send
(
sink_
,
atom
::
attach_v
,
self_
.
as_actor
());
std
::
string
str
=
"established new connection on socket "
;
str
+=
std
::
to_string
(
down
->
handle
().
id
);
self_
->
send
(
sink_
,
atom
::
status_v
,
std
::
move
(
str
));
return
{};
}
template
<
class
LowerLayerPtr
>
bool
prepare_send
(
LowerLayerPtr
down
)
{
// The lower level calls this function whenever a send event occurs, but
// before performing any socket I/O operation. We use this as a hook for
// constantly checking our mailbox. Returning `false` here aborts the
// application and closes the socket.
while
(
self_
->
consume_message
())
{
// We set abort_reason in our response handlers in case of an error.
if
(
down
->
abort_reason
())
return
false
;
// else: repeat.
}
return
true
;
}
template
<
class
LowerLayerPtr
>
bool
done_sending
(
LowerLayerPtr
down
)
{
// The lower level calls this function to check whether we are ready to
// unregister our socket from send events. We must make sure to put our
// mailbox in to the blocking state in order to get re-registered once new
// messages arrive.
return
self_
->
try_block_mailbox
();
}
template
<
class
LowerLayerPtr
>
void
abort
(
LowerLayerPtr
,
const
caf
::
error
&
reason
)
{
std
::
string
str
=
"app::abort called: "
;
str
+=
to_string
(
reason
);
self_
->
send
(
sink_
,
atom
::
status_v
,
std
::
move
(
str
));
self_
->
send
(
sink_
,
atom
::
detach_v
);
}
template
<
class
LowerLayerPtr
>
ptrdiff_t
consume_text
(
LowerLayerPtr
,
caf
::
string_view
text
)
{
self_
->
send
(
sink_
,
atom
::
text_v
,
std
::
string
{
text
.
begin
(),
text
.
end
()});
return
static_cast
<
ptrdiff_t
>
(
text
.
size
());
}
template
<
class
LowerLayerPtr
>
ptrdiff_t
consume_binary
(
LowerLayerPtr
,
caf
::
byte_span
bytes
)
{
self_
->
send
(
sink_
,
atom
::
binary_v
,
caf
::
byte_buffer
{
bytes
.
begin
(),
bytes
.
end
()});
return
static_cast
<
ptrdiff_t
>
(
bytes
.
size
());
}
private:
// Stores a handle to the actor that controls terminal output.
terminal_actor
sink_
;
// Enables the application to send and receive actor messages.
caf
::
net
::
actor_shell_ptr
self_
;
};
// -- implementation of the terminal actor -------------------------------------
struct
terminal_state
{
static
inline
const
char
*
name
=
"terminal"
;
terminal_actor
::
pointer
self
;
caf
::
actor
client
;
std
::
string
current_line
;
terminal_state
(
terminal_actor
::
pointer
self
)
:
self
(
self
)
{
// nop
}
void
render
(
const
char
*
str
)
{
puts
(
str
);
}
void
render
(
const
std
::
string
&
str
)
{
render
(
str
.
c_str
());
}
void
render
(
const
caf
::
byte_buffer
&
buf
)
{
for
(
auto
byte
:
buf
)
printf
(
"%02x"
,
caf
::
to_integer
<
int
>
(
byte
));
putc
(
'\n'
,
stdout
);
}
void
detach
()
{
puts
(
"*** server disconnected, bye!"
);
self
->
quit
();
}
template
<
class
T
>
void
render_server_message
(
const
char
*
prefix
,
const
T
&
msg
)
{
fputs
(
prefix
,
stdout
);
render
(
msg
);
}
terminal_actor
::
behavior_type
make_behavior
()
{
self
->
set_down_handler
([
this
](
const
caf
::
down_msg
&
dm
)
{
if
(
dm
.
source
==
client
)
detach
();
});
return
{
[
this
](
atom
::
attach
,
caf
::
actor
&
new_client
)
{
render_server_message
(
"[MSG] "
,
"client up and running"
);
client
=
std
::
move
(
new_client
);
self
->
monitor
(
client
);
},
[
this
](
atom
::
detach
)
{
if
(
self
->
current_sender
()
==
client
)
detach
();
},
[
this
](
atom
::
binary
,
const
caf
::
byte_buffer
&
buf
)
{
render_server_message
(
"[BIN] "
,
buf
);
},
[
this
](
atom
::
status
,
const
std
::
string
&
line
)
{
render_server_message
(
"[MSG] "
,
line
);
},
[
this
](
atom
::
text
,
const
std
::
string
&
line
)
{
render_server_message
(
"[TXT] "
,
line
);
},
[
this
](
atom
::
input
,
std
::
string
&
out
)
{
if
(
client
)
{
anon_send
(
client
,
std
::
move
(
out
));
}
else
{
puts
(
"[ERR] not connected to a server"
);
}
},
};
}
};
using
terminal_impl
=
terminal_actor
::
stateful_impl
<
terminal_state
>
;
// -- main ---------------------------------------------------------------------
struct
config
:
caf
::
actor_system_config
{
config
()
{
opt_group
{
custom_options_
,
"global"
}
//
.
add
<
caf
::
uri
>
(
"server,s"
,
"locator for the server, e.g., ws://localhost:8080"
)
.
add
<
std
::
string
>
(
"protocols"
,
"setting for the optional Sec-WebSocket-Protocol header field"
)
.
add
<
std
::
string
>
(
"extensions"
,
"setting for the optional Sec-WebSocket-Extensions header field"
);
}
};
int
main_loop
(
caf
::
scoped_actor
&
self
,
terminal_actor
term
)
{
std
::
string
line
;
while
(
std
::
getline
(
std
::
cin
,
line
))
{
line
.
push_back
(
'\n'
);
bool
fail
=
false
;
self
->
request
(
term
,
caf
::
infinite
,
atom
::
input_v
,
line
)
.
receive
([]
{
/* ok */
},
[
&
fail
](
const
caf
::
error
&
)
{
fail
=
true
;
});
if
(
fail
)
return
EXIT_FAILURE
;
}
return
EXIT_SUCCESS
;
}
int
caf_main
(
caf
::
actor_system
&
sys
,
const
config
&
cfg
)
{
using
namespace
caf
;
using
namespace
caf
::
net
;
auto
locator
=
get_or
(
cfg
,
"server"
,
uri
{});
if
(
locator
.
empty
())
{
std
::
cerr
<<
"*** mandatory option 'server' missing, use -h for help
\n
"
;
return
EXIT_FAILURE
;
}
else
if
(
locator
.
scheme
()
!=
"ws"
)
{
std
::
cerr
<<
"*** expected a WebSocket URI (ws://...), use -h for help
\n
"
;
return
EXIT_FAILURE
;
}
else
if
(
!
locator
.
fragment
().
empty
())
{
std
::
cerr
<<
"*** sorry: query and fragment components are not supported
\n
"
;
return
EXIT_FAILURE
;
}
else
if
(
locator
.
authority
().
empty
())
{
std
::
cerr
<<
"*** authority component missing in URI
\n
"
;
return
EXIT_FAILURE
;
}
else
if
(
auto
sock
=
make_connected_tcp_stream_socket
(
locator
.
authority
());
!
sock
)
{
std
::
cerr
<<
"*** failed to connect to "
<<
to_string
(
locator
.
authority
())
<<
": "
<<
to_string
(
sock
.
error
())
<<
'\n'
;
return
EXIT_FAILURE
;
}
else
{
// Fill the WebSocket handshake with the mandatory and optional fields.
web_socket
::
handshake
hs
;
hs
.
host
(
to_string
(
locator
.
authority
()));
if
(
locator
.
path
().
empty
())
hs
.
endpoint
(
"/"
);
else
hs
.
endpoint
(
to_string
(
locator
.
path
()));
if
(
auto
protocols
=
get_as
<
std
::
string
>
(
cfg
,
"protocols"
))
{
hs
.
protocols
(
std
::
move
(
*
protocols
));
}
if
(
auto
extensions
=
get_as
<
std
::
string
>
(
cfg
,
"extensions"
))
{
hs
.
extensions
(
std
::
move
(
*
extensions
));
}
// Spin up actors and enter main loop.
auto
term
=
sys
.
spawn
<
terminal_impl
>
();
auto
mgr
=
make_socket_manager
<
app
,
web_socket
::
client
,
stream_transport
>
(
*
sock
,
sys
.
network_manager
().
mpx_ptr
(),
std
::
move
(
hs
),
term
);
if
(
auto
err
=
mgr
->
init
(
content
(
cfg
)))
{
std
::
cerr
<<
"*** failed to initialize the WebSocket client: "
<<
to_string
(
err
)
<<
'\n'
;
return
EXIT_FAILURE
;
}
scoped_actor
self
{
sys
};
return
main_loop
(
self
,
term
);
}
}
CAF_MAIN
(
caf
::
id_block
::
web_socket_client
,
caf
::
net
::
middleman
)
examples/net/web-socket-feed.cpp
View file @
4d03a100
...
...
@@ -30,14 +30,14 @@
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/byte_span.hpp"
#include "caf/caf_main.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/exec_main.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/net/actor_shell.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/web_socket
_
server.hpp"
#include "caf/net/web_socket
/
server.hpp"
#include "caf/stateful_actor.hpp"
#include "caf/tag/mixed_message_oriented.hpp"
#include "caf/typed_event_based_actor.hpp"
...
...
@@ -164,7 +164,7 @@ class app {
public:
// -- member types -----------------------------------------------------------
//
We expect a stream-oriented interface to the lower communication layer
s.
//
Tells CAF we expect a transport with text and binary message
s.
using
input_tag
=
caf
::
tag
::
mixed_message_oriented
;
// -- constants --------------------------------------------------------------
...
...
@@ -289,7 +289,7 @@ int caf_main(caf::actor_system& sys, const config& cfg) {
// Spawn our feed actor and initiate the protocol stack.
stock
::
feed
feed
=
sys
.
spawn
(
feed_impl
);
auto
add_conn
=
[
feed
](
tcp_stream_socket
sock
,
multiplexer
*
mpx
)
{
return
make_socket_manager
<
app
,
web_socket
_
server
,
stream_transport
>
(
return
make_socket_manager
<
app
,
web_socket
::
server
,
stream_transport
>
(
sock
,
mpx
,
feed
);
};
sys
.
network_manager
().
make_acceptor
(
sock
,
add_conn
);
...
...
libcaf_net/CMakeLists.txt
View file @
4d03a100
...
...
@@ -38,6 +38,7 @@ caf_incubator_add_component(
src/net/middleman.cpp
src/net/middleman_backend.cpp
src/net/packet_writer.cpp
src/net/web_socket/handshake.cpp
src/network_socket.cpp
src/pipe_socket.cpp
src/pollset_updater.cpp
...
...
@@ -61,7 +62,9 @@ caf_incubator_add_component(
net.actor_shell
net.length_prefix_framing
net.typed_actor_shell
net.web_socket_server
net.web_socket.client
net.web_socket.handshake
net.web_socket.server
network_socket
pipe_socket
socket
...
...
libcaf_net/caf/net/basp/header.hpp
View file @
4d03a100
...
...
@@ -10,8 +10,6 @@
#include "caf/detail/comparable.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include "caf/meta/hex_formatted.hpp"
#include "caf/meta/type_name.hpp"
#include "caf/net/basp/constants.hpp"
#include "caf/net/basp/message_type.hpp"
#include "caf/type_id.hpp"
...
...
libcaf_net/caf/net/middleman.hpp
View file @
4d03a100
...
...
@@ -149,6 +149,14 @@ public:
return
mpx_
;
}
multiplexer
*
mpx_ptr
()
noexcept
{
return
&
mpx_
;
}
const
multiplexer
*
mpx_ptr
()
const
noexcept
{
return
&
mpx_
;
}
middleman_backend
*
backend
(
string_view
scheme
)
const
noexcept
;
expected
<
uint16_t
>
port
(
string_view
scheme
)
const
;
...
...
libcaf_net/caf/net/web_socket/client.hpp
0 → 100644
View file @
4d03a100
// 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 <algorithm>
#include "caf/byte_span.hpp"
#include "caf/detail/encode_base64.hpp"
#include "caf/error.hpp"
#include "caf/hash/sha1.hpp"
#include "caf/logger.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/net/web_socket/handshake.hpp"
#include "caf/net/web_socket/framing.hpp"
#include "caf/pec.hpp"
#include "caf/settings.hpp"
#include "caf/tag/mixed_message_oriented.hpp"
#include "caf/tag/stream_oriented.hpp"
namespace
caf
::
net
::
web_socket
{
/// Implements the client part for the WebSocket Protocol as defined in RFC
/// 6455. Initially, the layer performs the WebSocket handshake. Once completed,
/// this layer decodes RFC 6455 frames and forwards binary and text messages to
/// `UpperLayer`.
template
<
class
UpperLayer
>
class
client
{
public:
// -- member types -----------------------------------------------------------
using
input_tag
=
tag
::
stream_oriented
;
using
output_tag
=
tag
::
mixed_message_oriented
;
// -- constructors, destructors, and assignment operators --------------------
template
<
class
...
Ts
>
explicit
client
(
handshake
hs
,
Ts
&&
...
xs
)
:
upper_layer_
(
std
::
forward
<
Ts
>
(
xs
)...)
{
handshake_
.
reset
(
new
handshake
(
std
::
move
(
hs
)));
}
// -- initialization ---------------------------------------------------------
template
<
class
LowerLayerPtr
>
error
init
(
socket_manager
*
owner
,
LowerLayerPtr
down
,
const
settings
&
cfg
)
{
CAF_ASSERT
(
handshake_
!=
nullptr
);
owner_
=
owner
;
if
(
!
handshake_
->
has_mandatory_fields
())
return
make_error
(
sec
::
runtime_error
,
"handshake data lacks mandatory fields"
);
if
(
!
handshake_
->
has_valid_key
())
handshake_
->
randomize_key
();
cfg_
=
cfg
;
down
->
begin_output
();
handshake_
->
write_http_1_request
(
down
->
output_buffer
());
down
->
end_output
();
down
->
configure_read
(
receive_policy
::
up_to
(
handshake
::
max_http_size
));
return
none
;
}
// -- properties -------------------------------------------------------------
auto
&
upper_layer
()
noexcept
{
return
upper_layer_
.
upper_layer
();
}
const
auto
&
upper_layer
()
const
noexcept
{
return
upper_layer_
.
upper_layer
();
}
// -- role: upper layer ------------------------------------------------------
template
<
class
LowerLayerPtr
>
bool
prepare_send
(
LowerLayerPtr
down
)
{
return
handshake_complete
()
?
upper_layer_
.
prepare_send
(
down
)
:
true
;
}
template
<
class
LowerLayerPtr
>
bool
done_sending
(
LowerLayerPtr
down
)
{
return
handshake_complete
()
?
upper_layer_
.
done_sending
(
down
)
:
true
;
}
template
<
class
LowerLayerPtr
>
void
abort
(
LowerLayerPtr
down
,
const
error
&
reason
)
{
if
(
handshake_complete
())
upper_layer_
.
abort
(
down
,
reason
);
}
template
<
class
LowerLayerPtr
>
ptrdiff_t
consume
(
LowerLayerPtr
down
,
byte_span
input
,
byte_span
delta
)
{
CAF_LOG_TRACE
(
CAF_ARG2
(
"socket"
,
down
->
handle
().
id
)
<<
CAF_ARG2
(
"bytes"
,
input
.
size
()));
// Short circuit to the framing layer after the handshake completed.
if
(
handshake_complete
())
return
upper_layer_
.
consume
(
down
,
input
,
delta
);
// Check whether received a HTTP header or else wait for more data or abort
// when exceeding the maximum size.
auto
[
hdr
,
remainder
]
=
handshake
::
split_http_1_header
(
input
);
if
(
hdr
.
empty
())
{
if
(
input
.
size
()
>=
handshake
::
max_http_size
)
{
down
->
begin_output
();
handshake
::
write_http_1_header_too_large
(
down
->
output_buffer
());
down
->
end_output
();
auto
err
=
make_error
(
pec
::
too_many_characters
,
"exceeded maximum header size"
);
down
->
abort_reason
(
std
::
move
(
err
));
return
-
1
;
}
else
{
return
0
;
}
}
else
if
(
!
handle_header
(
down
,
hdr
))
{
return
-
1
;
}
else
if
(
remainder
.
empty
())
{
CAF_ASSERT
(
hdr
.
size
()
==
input
.
size
());
return
hdr
.
size
();
}
else
{
CAF_LOG_DEBUG
(
CAF_ARG2
(
"socket"
,
down
->
handle
().
id
)
<<
CAF_ARG2
(
"remainder"
,
remainder
.
size
()));
if
(
auto
sub_result
=
upper_layer_
.
consume
(
down
,
remainder
,
remainder
);
sub_result
>=
0
)
{
return
hdr
.
size
()
+
sub_result
;
}
else
{
return
sub_result
;
}
}
}
bool
handshake_complete
()
const
noexcept
{
return
handshake_
==
nullptr
;
}
private:
// -- HTTP response processing -----------------------------------------------
template
<
class
LowerLayerPtr
>
bool
handle_header
(
LowerLayerPtr
down
,
string_view
http
)
{
CAF_ASSERT
(
handshake_
!=
nullptr
);
auto
http_ok
=
handshake_
->
is_valid_http_1_response
(
http
);
handshake_
.
reset
();
if
(
http_ok
)
{
if
(
auto
err
=
upper_layer_
.
init
(
owner_
,
down
,
cfg_
))
{
CAF_LOG_DEBUG
(
"failed to initialize WebSocket framing layer"
);
down
->
abort_reason
(
std
::
move
(
err
));
return
false
;
}
else
{
CAF_LOG_DEBUG
(
"completed WebSocket handshake"
);
return
true
;
}
}
else
{
CAF_LOG_DEBUG
(
"received invalid WebSocket handshake"
);
down
->
abort_reason
(
make_error
(
sec
::
runtime_error
,
"received invalid WebSocket handshake response from server"
));
return
false
;
}
}
// -- member variables -------------------------------------------------------
/// Stores the WebSocket handshake data until the handshake completed.
std
::
unique_ptr
<
handshake
>
handshake_
;
/// Stores the upper layer.
framing
<
UpperLayer
>
upper_layer_
;
/// Stores a pointer to the owning manager for the delayed initialization.
socket_manager
*
owner_
=
nullptr
;
settings
cfg_
;
};
}
// namespace caf::net::web_socket
libcaf_net/caf/net/web_socket/framing.hpp
0 → 100644
View file @
4d03a100
This diff is collapsed.
Click to expand it.
libcaf_net/caf/net/web_socket/handshake.hpp
0 → 100644
View file @
4d03a100
// 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 <string>
#include "caf/byte.hpp"
#include "caf/byte_buffer.hpp"
#include "caf/dictionary.hpp"
#include "caf/string_view.hpp"
namespace
caf
::
net
::
web_socket
{
/// Wraps state and algorithms for the WebSocket client handshake as defined in
/// RFC 6455.
class
handshake
{
public:
// -- member types -----------------------------------------------------------
using
key_type
=
std
::
array
<
byte
,
16
>
;
// -- constants --------------------------------------------------------------
/// Maximum size for HTTP request and response messages. A handshake should
/// usually fit into 200-300 Bytes, so 2KB is more than enough.
static
constexpr
uint32_t
max_http_size
=
2048
;
// -- constructors, destructors, and assignment operators --------------------
handshake
()
noexcept
;
handshake
(
const
handshake
&
)
=
default
;
handshake
(
handshake
&&
)
noexcept
=
default
;
handshake
&
operator
=
(
const
handshake
&
)
=
default
;
handshake
&
operator
=
(
handshake
&&
)
noexcept
=
default
;
// -- properties -------------------------------------------------------------
/// Returns the 16-byte `key` for the opening handshake.
[[
nodiscard
]]
const
auto
&
key
()
const
noexcept
{
return
key_
;
}
/// Sets the 16-byte `key` for the opening handshake.
void
key
(
key_type
new_key
)
noexcept
{
key_
=
new_key
;
}
/// Tries to assign a key from base64 input.
bool
assign_key
(
string_view
base64_key
);
/// Returns the key for the response message, i.e., what the server puts into
/// the HTTP field `Sec-WebSocket-Accept`.
std
::
string
response_key
()
const
;
/// Returns all configured header fields, except the key.
[[
nodiscard
]]
const
auto
&
fields
()
const
noexcept
{
return
fields_
;
}
/// Checks whether at least one bit in the key is one. A default constructed
/// header object fills the key with zeros.
[[
nodiscard
]]
bool
has_valid_key
()
const
noexcept
;
/// Checks whether all mandatory fields were provided by the user.
[[
nodiscard
]]
bool
has_mandatory_fields
()
const
noexcept
;
/// Fills the key with pseudo-random numbers generated by `std::minstd_rand`
/// with a seed chosen from `std::random_device`.
void
randomize_key
();
/// Fills the key with pseudo-random numbers generated by `std::minstd_rand`,
/// initialized with `seed`.
void
randomize_key
(
unsigned
seed
);
/// Sets a value for the mandatory WebSocket endpoint, i.e., the Request-URI
/// component of the GET method according to RFC 2616.
/// @param value Identifies the endpoint that should handle the request.
void
endpoint
(
std
::
string
value
)
{
fields_
[
"_endpoint"
]
=
std
::
move
(
value
);
}
/// Sets a value for the mandatory `Host` field.
/// @param value The Internet host and port number of the resource being
/// requested, as obtained from the original URI given by the
/// user or referring resource.
void
host
(
std
::
string
value
)
{
fields_
[
"_host"
]
=
std
::
move
(
value
);
}
/// Sets a value for the optional `Origin` field.
/// @param value Indicates where the GET request originates from. Usually only
/// sent by browser clients.
void
origin
(
std
::
string
value
)
{
fields_
[
"Origin"
]
=
std
::
move
(
value
);
}
/// Sets a value for the optional `Sec-WebSocket-Protocol` field.
/// @param value A comma-separated list of values indicating which protocols
/// the client would like to speak, ordered by preference.
void
protocols
(
std
::
string
value
)
{
fields_
[
"Sec-WebSocket-Protocol"
]
=
std
::
move
(
value
);
}
/// Sets a value for the optional `Sec-WebSocket-Extensions` field.
/// @param value A comma-separated list of values indicating which extensions
/// the client would like to speak, ordered by preference.
void
extensions
(
std
::
string
value
)
{
fields_
[
"Sec-WebSocket-Extensions"
]
=
std
::
move
(
value
);
}
// -- HTTP generation and validation -----------------------------------------
/// Writes the HTTP 1.1 request message to `buf`.
/// @pre `has_mandatory_fields()`
void
write_http_1_request
(
byte_buffer
&
buf
)
const
;
/// Writes the HTTP 1.1 response message to `buf`.
/// @pre `has_valid_key()`
void
write_http_1_response
(
byte_buffer
&
buf
)
const
;
/// Writes a HTTP 1.1 431 (Request Header Fields Too Large) response.
static
void
write_http_1_header_too_large
(
byte_buffer
&
buf
);
/// Checks whether the `http_response` contains a HTTP 1.1 response to the
/// generated HTTP GET request. A valid response contains:
/// - HTTP status code 101 (Switching Protocols).
/// - An `Upgrade` field with the value `websocket`.
/// - A `Connection` field with the value `Upgrade`.
/// - A `Sec-WebSocket-Accept` field with the value `response_key()`.
bool
is_valid_http_1_response
(
string_view
http_response
)
const
;
/// Tries splitting the given byte span into an HTTP header (`first`) and a
/// remainder (`second`). Returns an empty @ref string_view as `first` for
/// incomplete HTTP headers.
static
std
::
pair
<
string_view
,
byte_span
>
split_http_1_header
(
byte_span
bytes
);
private:
// -- utility ----------------------------------------------------------------
string_view
lookup
(
string_view
field_name
)
const
noexcept
;
// -- member variables -------------------------------------------------------
key_type
key_
;
dictionary
<
std
::
string
>
fields_
;
};
}
// namespace caf::net::web_socket
libcaf_net/caf/net/web_socket/server.hpp
0 → 100644
View file @
4d03a100
// 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 <algorithm>
#include "caf/byte_span.hpp"
#include "caf/error.hpp"
#include "caf/logger.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/net/web_socket/framing.hpp"
#include "caf/net/web_socket/handshake.hpp"
#include "caf/pec.hpp"
#include "caf/settings.hpp"
#include "caf/tag/mixed_message_oriented.hpp"
#include "caf/tag/stream_oriented.hpp"
namespace
caf
::
net
::
web_socket
{
/// Implements the server part for the WebSocket Protocol as defined in RFC
/// 6455. Initially, the layer performs the WebSocket handshake. Once completed,
/// this layer decodes RFC 6455 frames and forwards binary and text messages to
/// `UpperLayer`.
template
<
class
UpperLayer
>
class
server
{
public:
// -- member types -----------------------------------------------------------
using
input_tag
=
tag
::
stream_oriented
;
using
output_tag
=
tag
::
mixed_message_oriented
;
// -- constructors, destructors, and assignment operators --------------------
template
<
class
...
Ts
>
explicit
server
(
Ts
&&
...
xs
)
:
upper_layer_
(
std
::
forward
<
Ts
>
(
xs
)...)
{
// > A server MUST NOT mask any frames that it sends to the client.
// See RFC 6455, Section 5.1.
upper_layer_
.
mask_outgoing_frames
=
false
;
}
// -- initialization ---------------------------------------------------------
template
<
class
LowerLayerPtr
>
error
init
(
socket_manager
*
owner
,
LowerLayerPtr
down
,
const
settings
&
cfg
)
{
owner_
=
owner
;
cfg_
=
cfg
;
down
->
configure_read
(
receive_policy
::
up_to
(
handshake
::
max_http_size
));
return
none
;
}
// -- properties -------------------------------------------------------------
auto
&
upper_layer
()
noexcept
{
return
upper_layer_
.
upper_layer
();
}
const
auto
&
upper_layer
()
const
noexcept
{
return
upper_layer_
.
upper_layer
();
}
// -- role: upper layer ------------------------------------------------------
template
<
class
LowerLayerPtr
>
bool
prepare_send
(
LowerLayerPtr
down
)
{
return
handshake_complete_
?
upper_layer_
.
prepare_send
(
down
)
:
true
;
}
template
<
class
LowerLayerPtr
>
bool
done_sending
(
LowerLayerPtr
down
)
{
return
handshake_complete_
?
upper_layer_
.
done_sending
(
down
)
:
true
;
}
template
<
class
LowerLayerPtr
>
void
abort
(
LowerLayerPtr
down
,
const
error
&
reason
)
{
if
(
handshake_complete_
)
upper_layer_
.
abort
(
down
,
reason
);
}
template
<
class
LowerLayerPtr
>
ptrdiff_t
consume
(
LowerLayerPtr
down
,
byte_span
input
,
byte_span
delta
)
{
CAF_LOG_TRACE
(
CAF_ARG2
(
"socket"
,
down
->
handle
().
id
)
<<
CAF_ARG2
(
"bytes"
,
input
.
size
()));
// Short circuit to the framing layer after the handshake completed.
if
(
handshake_complete_
)
return
upper_layer_
.
consume
(
down
,
input
,
delta
);
// Check whether received a HTTP header or else wait for more data or abort
// when exceeding the maximum size.
auto
[
hdr
,
remainder
]
=
handshake
::
split_http_1_header
(
input
);
if
(
hdr
.
empty
())
{
if
(
input
.
size
()
>=
handshake
::
max_http_size
)
{
down
->
begin_output
();
handshake
::
write_http_1_header_too_large
(
down
->
output_buffer
());
down
->
end_output
();
auto
err
=
make_error
(
pec
::
too_many_characters
,
"exceeded maximum header size"
);
down
->
abort_reason
(
std
::
move
(
err
));
return
-
1
;
}
else
{
return
0
;
}
}
else
if
(
!
handle_header
(
down
,
hdr
))
{
return
-
1
;
}
else
if
(
remainder
.
empty
())
{
CAF_ASSERT
(
hdr
.
size
()
==
input
.
size
());
return
hdr
.
size
();
}
else
{
CAF_LOG_DEBUG
(
CAF_ARG2
(
"socket"
,
down
->
handle
().
id
)
<<
CAF_ARG2
(
"remainder"
,
remainder
.
size
()));
if
(
auto
sub_result
=
upper_layer_
.
consume
(
down
,
remainder
,
remainder
);
sub_result
>=
0
)
{
return
hdr
.
size
()
+
sub_result
;
}
else
{
return
sub_result
;
}
}
}
bool
handshake_complete
()
const
noexcept
{
return
handshake_complete_
;
}
private:
// -- HTTP request processing ------------------------------------------------
template
<
class
LowerLayerPtr
>
bool
handle_header
(
LowerLayerPtr
down
,
string_view
http
)
{
// Parse the first line, i.e., "METHOD REQUEST-URI VERSION".
auto
[
first_line
,
remainder
]
=
split
(
http
,
"
\r\n
"
);
auto
[
method
,
request_uri
,
version
]
=
split2
(
first_line
,
" "
);
auto
&
hdr
=
cfg_
[
"web-socket"
].
as_dictionary
();
if
(
method
!=
"GET"
)
{
auto
err
=
make_error
(
pec
::
invalid_argument
,
"invalid operation: expected GET, got "
+
to_string
(
method
));
down
->
abort_reason
(
std
::
move
(
err
));
return
false
;
}
// Store the request information in the settings for the upper layer.
put
(
hdr
,
"method"
,
method
);
put
(
hdr
,
"request-uri"
,
request_uri
);
put
(
hdr
,
"http-version"
,
version
);
// Store the remaining header fields.
auto
&
fields
=
hdr
[
"fields"
].
as_dictionary
();
for_each_line
(
remainder
,
[
&
fields
](
string_view
line
)
{
if
(
auto
sep
=
std
::
find
(
line
.
begin
(),
line
.
end
(),
':'
);
sep
!=
line
.
end
())
{
auto
key
=
trim
({
line
.
begin
(),
sep
});
auto
val
=
trim
({
sep
+
1
,
line
.
end
()});
if
(
!
key
.
empty
())
put
(
fields
,
key
,
val
);
}
});
// Check whether the mandatory fields exist.
handshake
hs
;
if
(
auto
skey_field
=
get_if
<
std
::
string
>
(
&
fields
,
"Sec-WebSocket-Key"
);
skey_field
&&
hs
.
assign_key
(
*
skey_field
))
{
CAF_LOG_DEBUG
(
"received Sec-WebSocket-Key"
<<
*
skey_field
);
}
else
{
CAF_LOG_DEBUG
(
"received invalid WebSocket handshake"
);
down
->
abort_reason
(
make_error
(
pec
::
missing_field
,
"mandatory field Sec-WebSocket-Key missing or invalid"
));
return
false
;
}
// Send server handshake.
down
->
begin_output
();
hs
.
write_http_1_response
(
down
->
output_buffer
());
down
->
end_output
();
// Try initializing the upper layer.
if
(
auto
err
=
upper_layer_
.
init
(
owner_
,
down
,
cfg_
))
{
down
->
abort_reason
(
std
::
move
(
err
));
return
false
;
}
// Done.
CAF_LOG_DEBUG
(
"completed WebSocket handshake"
);
handshake_complete_
=
true
;
return
true
;
}
template
<
class
F
>
void
for_each_line
(
string_view
input
,
F
&&
f
)
{
constexpr
string_view
eol
{
"
\r\n
"
};
for
(
auto
pos
=
input
.
begin
();;)
{
auto
line_end
=
std
::
search
(
pos
,
input
.
end
(),
eol
.
begin
(),
eol
.
end
());
if
(
line_end
==
input
.
end
())
return
;
f
(
string_view
{
pos
,
line_end
});
pos
=
line_end
+
eol
.
size
();
}
}
static
string_view
trim
(
string_view
str
)
{
str
.
remove_prefix
(
std
::
min
(
str
.
find_first_not_of
(
' '
),
str
.
size
()));
auto
trim_pos
=
str
.
find_last_not_of
(
' '
);
if
(
trim_pos
!=
str
.
npos
)
str
.
remove_suffix
(
str
.
size
()
-
(
trim_pos
+
1
));
return
str
;
}
/// Splits `str` at the first occurrence of `sep` into the head and the
/// remainder (excluding the separator).
static
std
::
pair
<
string_view
,
string_view
>
split
(
string_view
str
,
string_view
sep
)
{
auto
i
=
std
::
search
(
str
.
begin
(),
str
.
end
(),
sep
.
begin
(),
sep
.
end
());
if
(
i
!=
str
.
end
())
return
{{
str
.
begin
(),
i
},
{
i
+
sep
.
size
(),
str
.
end
()}};
return
{{
str
},
{}};
}
/// Convenience function for splitting twice.
static
std
::
tuple
<
string_view
,
string_view
,
string_view
>
split2
(
string_view
str
,
string_view
sep
)
{
auto
[
first
,
r1
]
=
split
(
str
,
sep
);
auto
[
second
,
third
]
=
split
(
r1
,
sep
);
return
{
first
,
second
,
third
};
}
/// Stores whether the WebSocket handshake completed successfully.
bool
handshake_complete_
=
false
;
/// Stores the upper layer.
framing
<
UpperLayer
>
upper_layer_
;
/// Stores a pointer to the owning manager for the delayed initialization.
socket_manager
*
owner_
=
nullptr
;
/// Holds a copy of the settings in order to delay initialization of the upper
/// layer until the handshake completed.
settings
cfg_
;
};
}
// namespace caf::net::web_socket
libcaf_net/caf/net/web_socket_framing.hpp
View file @
4d03a100
This diff is collapsed.
Click to expand it.
libcaf_net/caf/net/web_socket_server.hpp
View file @
4d03a100
...
...
@@ -4,272 +4,13 @@
#pragma once
#include <algorithm>
#include "caf/byte_span.hpp"
#include "caf/detail/encode_base64.hpp"
#include "caf/error.hpp"
#include "caf/hash/sha1.hpp"
#include "caf/logger.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/net/web_socket_framing.hpp"
#include "caf/pec.hpp"
#include "caf/settings.hpp"
#include "caf/tag/mixed_message_oriented.hpp"
#include "caf/tag/stream_oriented.hpp"
#include "caf/net/web_socket/server.hpp"
namespace
caf
::
net
{
/// Implements the WebSocket Protocol as defined in RFC 6455. Initially, the
/// layer performs the WebSocket handshake. Once completed, this layer decodes
/// RFC 6455 frames and forwards binary and text messages to `UpperLayer`.
template
<
class
UpperLayer
>
class
web_socket_server
{
public:
// -- member types -----------------------------------------------------------
using
input_tag
=
tag
::
stream_oriented
;
using
output_tag
=
tag
::
mixed_message_oriented
;
// -- constants --------------------------------------------------------------
static
constexpr
std
::
array
<
byte
,
4
>
end_of_header
{{
byte
{
'\r'
},
byte
{
'\n'
},
byte
{
'\r'
},
byte
{
'\n'
},
}};
// A handshake should usually fit into 200-300 Bytes. 2KB is more than enough.
static
constexpr
uint32_t
max_header_size
=
2048
;
static
constexpr
string_view
header_too_large
=
"HTTP/1.1 431 Request Header Fields Too Large
\r\n
"
"Content-Type: text/plain
\r\n
"
"
\r\n
"
"Header exceeds 2048 Bytes.
\r\n
"
;
// -- constructors, destructors, and assignment operators --------------------
template
<
class
...
Ts
>
explicit
web_socket_server
(
Ts
&&
...
xs
)
:
upper_layer_
(
std
::
forward
<
Ts
>
(
xs
)...)
{
// > A server MUST NOT mask any frames that it sends to the client.
// See RFC 6455, Section 5.1.
upper_layer_
.
mask_outgoing_frames
=
false
;
}
// -- initialization ---------------------------------------------------------
template
<
class
LowerLayerPtr
>
error
init
(
socket_manager
*
owner
,
LowerLayerPtr
down
,
const
settings
&
cfg
)
{
owner_
=
owner
;
cfg_
=
cfg
;
down
->
configure_read
(
net
::
receive_policy
::
up_to
(
max_header_size
));
return
none
;
}
// -- properties -------------------------------------------------------------
auto
&
upper_layer
()
noexcept
{
return
upper_layer_
.
upper_layer
();
}
const
auto
&
upper_layer
()
const
noexcept
{
return
upper_layer_
.
upper_layer
();
}
// -- role: upper layer ------------------------------------------------------
template
<
class
LowerLayerPtr
>
bool
prepare_send
(
LowerLayerPtr
down
)
{
return
handshake_complete_
?
upper_layer_
.
prepare_send
(
down
)
:
true
;
}
template
<
class
LowerLayerPtr
>
bool
done_sending
(
LowerLayerPtr
down
)
{
return
handshake_complete_
?
upper_layer_
.
done_sending
(
down
)
:
true
;
}
template
<
class
LowerLayerPtr
>
void
abort
(
LowerLayerPtr
down
,
const
error
&
reason
)
{
if
(
handshake_complete_
)
upper_layer_
.
abort
(
down
,
reason
);
}
template
<
class
LowerLayerPtr
>
ptrdiff_t
consume
(
LowerLayerPtr
down
,
byte_span
buffer
,
byte_span
delta
)
{
CAF_LOG_TRACE
(
CAF_ARG2
(
"socket"
,
down
->
handle
().
id
)
<<
CAF_ARG2
(
"bytes"
,
buffer
.
size
())
<<
CAF_ARG2
(
"delta"
,
delta
.
size
()));
if
(
handshake_complete_
)
return
upper_layer_
.
consume
(
down
,
buffer
,
delta
);
// TODO: we could avoid repeated scans by using the delta parameter.
auto
i
=
std
::
search
(
buffer
.
begin
(),
buffer
.
end
(),
end_of_header
.
begin
(),
end_of_header
.
end
());
if
(
i
==
buffer
.
end
())
{
if
(
buffer
.
size
()
==
max_header_size
)
{
write
(
down
,
header_too_large
);
auto
err
=
make_error
(
pec
::
too_many_characters
,
"exceeded maximum header size"
);
down
->
abort_reason
(
std
::
move
(
err
));
return
-
1
;
}
return
0
;
}
auto
offset
=
static_cast
<
size_t
>
(
std
::
distance
(
buffer
.
begin
(),
i
));
offset
+=
end_of_header
.
size
();
// Take all but the last two bytes (to avoid an empty line) as input for
// the header.
string_view
header
{
reinterpret_cast
<
const
char
*>
(
buffer
.
begin
()),
offset
-
2
};
if
(
!
handle_header
(
down
,
header
))
return
-
1
;
ptrdiff_t
sub_result
=
0
;
if
(
offset
<
buffer
.
size
())
{
auto
remainder
=
buffer
.
subspan
(
offset
);
CAF_LOG_TRACE
(
CAF_ARG2
(
"socket"
,
down
->
handle
().
id
)
<<
CAF_ARG2
(
"remainder"
,
remainder
.
size
()));
sub_result
=
upper_layer_
.
consume
(
down
,
remainder
,
remainder
);
if
(
sub_result
<
0
)
return
sub_result
;
}
return
static_cast
<
ptrdiff_t
>
(
offset
)
+
sub_result
;
}
bool
handshake_complete
()
const
noexcept
{
return
handshake_complete_
;
}
private:
template
<
class
LowerLayerPtr
>
static
void
write
(
LowerLayerPtr
down
,
string_view
output
)
{
auto
out
=
as_bytes
(
make_span
(
output
));
down
->
begin_output
();
auto
&
buf
=
down
->
output_buffer
();
buf
.
insert
(
buf
.
end
(),
out
.
begin
(),
out
.
end
());
down
->
end_output
();
}
template
<
class
LowerLayerPtr
>
bool
handle_header
(
LowerLayerPtr
down
,
string_view
input
)
{
// Parse the first line, i.e., "METHOD REQUEST-URI VERSION".
auto
[
first_line
,
remainder
]
=
split
(
input
,
"
\r\n
"
);
auto
[
method
,
request_uri
,
version
]
=
split2
(
first_line
,
" "
);
auto
&
hdr
=
cfg_
[
"web-socket"
].
as_dictionary
();
if
(
method
!=
"GET"
)
{
auto
err
=
make_error
(
pec
::
invalid_argument
,
"invalid operation: expected GET, got "
+
to_string
(
method
));
down
->
abort_reason
(
std
::
move
(
err
));
return
false
;
}
// Store the request information in the settings for the upper layer.
put
(
hdr
,
"method"
,
method
);
put
(
hdr
,
"request-uri"
,
request_uri
);
put
(
hdr
,
"http-version"
,
version
);
// Store the remaining header fields.
auto
&
fields
=
hdr
[
"fields"
].
as_dictionary
();
for_each_line
(
remainder
,
[
&
fields
](
string_view
line
)
{
if
(
auto
sep
=
std
::
find
(
line
.
begin
(),
line
.
end
(),
':'
);
sep
!=
line
.
end
())
{
auto
key
=
trim
({
line
.
begin
(),
sep
});
auto
val
=
trim
({
sep
+
1
,
line
.
end
()});
if
(
!
key
.
empty
())
put
(
fields
,
key
,
val
);
}
});
// Check whether the mandatory fields exist.
std
::
string
sec_key
;
if
(
auto
skey_field
=
get_if
<
std
::
string
>
(
&
fields
,
"Sec-WebSocket-Key"
))
{
auto
magic
=
*
skey_field
+
"258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
;
auto
field_hash
=
hash
::
sha1
::
compute
(
magic
);
sec_key
=
detail
::
encode_base64
(
field_hash
);
CAF_LOG_DEBUG
(
"received Sec-WebSocket-Key"
<<
*
skey_field
<<
"respond with"
<<
sec_key
);
}
else
{
auto
err
=
make_error
(
pec
::
missing_field
,
"Mandatory field Sec-WebSocket-Key not found"
);
down
->
abort_reason
(
std
::
move
(
err
));
return
false
;
}
// Send server handshake.
down
->
begin_output
();
auto
&
buf
=
down
->
output_buffer
();
auto
append
=
[
&
buf
](
string_view
output
)
{
auto
out
=
as_bytes
(
make_span
(
output
));
buf
.
insert
(
buf
.
end
(),
out
.
begin
(),
out
.
end
());
};
append
(
"HTTP/1.1 101 Switching Protocols
\r\n
"
"Upgrade: websocket
\r\n
"
"Connection: Upgrade
\r\n
"
"Sec-WebSocket-Accept: "
);
append
(
sec_key
);
append
(
"
\r\n\r\n
"
);
down
->
end_output
();
// Try initializing the upper layer.
if
(
auto
err
=
upper_layer_
.
init
(
owner_
,
down
,
cfg_
))
{
down
->
abort_reason
(
std
::
move
(
err
));
return
false
;
}
// Done.
CAF_LOG_DEBUG
(
"handshake completed"
);
handshake_complete_
=
true
;
return
true
;
}
template
<
class
F
>
void
for_each_line
(
string_view
input
,
F
&&
f
)
{
constexpr
string_view
eol
{
"
\r\n
"
};
for
(
auto
pos
=
input
.
begin
();;)
{
auto
line_end
=
std
::
search
(
pos
,
input
.
end
(),
eol
.
begin
(),
eol
.
end
());
if
(
line_end
==
input
.
end
())
return
;
f
(
string_view
{
pos
,
line_end
});
pos
=
line_end
+
eol
.
size
();
}
}
static
string_view
trim
(
string_view
str
)
{
str
.
remove_prefix
(
std
::
min
(
str
.
find_first_not_of
(
' '
),
str
.
size
()));
auto
trim_pos
=
str
.
find_last_not_of
(
' '
);
if
(
trim_pos
!=
str
.
npos
)
str
.
remove_suffix
(
str
.
size
()
-
(
trim_pos
+
1
));
return
str
;
}
/// Splits `str` at the first occurrence of `sep` into the head and the
/// remainder (excluding the separator).
static
std
::
pair
<
string_view
,
string_view
>
split
(
string_view
str
,
string_view
sep
)
{
auto
i
=
std
::
search
(
str
.
begin
(),
str
.
end
(),
sep
.
begin
(),
sep
.
end
());
if
(
i
!=
str
.
end
())
return
{{
str
.
begin
(),
i
},
{
i
+
sep
.
size
(),
str
.
end
()}};
return
{{
str
},
{}};
}
/// Convenience function for splitting twice.
static
std
::
tuple
<
string_view
,
string_view
,
string_view
>
split2
(
string_view
str
,
string_view
sep
)
{
auto
[
first
,
r1
]
=
split
(
str
,
sep
);
auto
[
second
,
third
]
=
split
(
r1
,
sep
);
return
{
first
,
second
,
third
};
}
/// Stores whether the WebSocket handshake completed successfully.
bool
handshake_complete_
=
false
;
/// Stores the upper layer.
web_socket_framing
<
UpperLayer
>
upper_layer_
;
/// Stores a pointer to the owning manager for the delayed initialization.
socket_manager
*
owner_
=
nullptr
;
/// Holds a copy of the settings in order to delay initialization of the upper
/// layer until the handshake completed.
settings
cfg_
;
};
using
web_socket_server
[[
deprecated
(
"use caf::net::web_socket::server instead"
)]]
=
web_socket
::
server
<
UpperLayer
>
;
}
// namespace caf::net
libcaf_net/src/net/web_socket/handshake.cpp
0 → 100644
View file @
4d03a100
// 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/handshake.hpp"
#include <algorithm>
#include <cctype>
#include <cstring>
#include <random>
#include <tuple>
#include "caf/config.hpp"
#include "caf/detail/encode_base64.hpp"
#include "caf/hash/sha1.hpp"
#include "caf/string_algorithms.hpp"
namespace
caf
::
net
::
web_socket
{
handshake
::
handshake
()
noexcept
{
key_
.
fill
(
byte
{
0
});
}
bool
handshake
::
has_valid_key
()
const
noexcept
{
auto
non_zero
=
[](
byte
x
)
{
return
x
!=
byte
{
0
};
};
return
std
::
any_of
(
key_
.
begin
(),
key_
.
end
(),
non_zero
);
}
bool
handshake
::
assign_key
(
string_view
base64_key
)
{
// Base 64 produces character groups of size 4. This means our key has to use
// six groups, but the last two characters are always padding.
if
(
base64_key
.
size
()
==
24
&&
ends_with
(
base64_key
,
"=="
))
{
std
::
vector
<
byte
>
buf
;
buf
.
reserve
(
18
);
if
(
detail
::
base64
::
decode
(
base64_key
,
buf
))
{
CAF_ASSERT
(
buf
.
size
()
==
16
);
key_type
bytes
;
std
::
copy
(
buf
.
begin
(),
buf
.
end
(),
bytes
.
begin
());
key
(
bytes
);
return
true
;
}
}
return
false
;
}
std
::
string
handshake
::
response_key
()
const
{
// For details on the (convoluted) algorithm see RFC 6455.
auto
str
=
detail
::
base64
::
encode
(
key_
);
str
+=
"258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
;
auto
sha
=
hash
::
sha1
::
compute
(
str
);
str
.
clear
();
detail
::
base64
::
encode
(
sha
,
str
);
return
str
;
}
void
handshake
::
randomize_key
()
{
std
::
random_device
rd
;
randomize_key
(
rd
());
}
void
handshake
::
randomize_key
(
unsigned
seed
)
{
std
::
minstd_rand
rng
{
seed
};
std
::
uniform_int_distribution
<>
f
{
0
,
255
};
for
(
auto
&
x
:
key_
)
x
=
static_cast
<
byte
>
(
f
(
rng
));
}
bool
handshake
::
has_mandatory_fields
()
const
noexcept
{
return
fields_
.
contains
(
"_endpoint"
)
&&
fields_
.
contains
(
"_host"
);
}
// -- HTTP generation and validation -------------------------------------------
namespace
{
struct
writer
{
byte_buffer
*
buf
;
};
writer
&
operator
<<
(
writer
&
out
,
string_view
str
)
{
auto
bytes
=
as_bytes
(
make_span
(
str
));
out
.
buf
->
insert
(
out
.
buf
->
end
(),
bytes
.
begin
(),
bytes
.
end
());
return
out
;
}
template
<
class
F
>
auto
operator
<<
(
writer
&
out
,
F
&&
f
)
->
decltype
(
f
(
out
))
{
return
f
(
out
);
}
}
// namespace
void
handshake
::
write_http_1_request
(
byte_buffer
&
buf
)
const
{
auto
encoded_key
=
[
this
](
auto
&
out
)
->
decltype
(
auto
)
{
detail
::
base64
::
encode
(
key_
,
*
out
.
buf
);
return
out
;
};
writer
out
{
&
buf
};
out
<<
"GET "
<<
lookup
(
"_endpoint"
)
<<
" HTTP/1.1
\r\n
"
<<
"Host: "
<<
lookup
(
"_host"
)
<<
"
\r\n
"
<<
"Upgrade: websocket
\r\n
"
<<
"Connection: Upgrade
\r\n
"
<<
"Sec-WebSocket-Version: 13
\r\n
"
<<
"Sec-WebSocket-Key: "
<<
encoded_key
<<
"
\r\n
"
;
for
(
auto
&
[
key
,
val
]
:
fields_
)
if
(
key
[
0
]
!=
'_'
)
out
<<
key
<<
": "
<<
val
<<
"
\r\n
"
;
out
<<
"
\r\n
"
;
}
void
handshake
::
write_http_1_response
(
byte_buffer
&
buf
)
const
{
writer
out
{
&
buf
};
out
<<
"HTTP/1.1 101 Switching Protocols
\r\n
"
"Upgrade: websocket
\r\n
"
"Connection: Upgrade
\r\n
"
"Sec-WebSocket-Accept: "
<<
response_key
()
<<
"
\r\n\r\n
"
;
}
void
handshake
::
write_http_1_header_too_large
(
byte_buffer
&
buf
)
{
writer
out
{
&
buf
};
out
<<
"HTTP/1.1 431 Request Header Fields Too Large
\r\n
"
"Content-Type: text/plain
\r\n
"
"
\r\n
"
"Header exceeds 2048 Bytes.
\r\n
"
;
}
namespace
{
template
<
class
F
>
void
for_each_http_line
(
string_view
lines
,
F
&&
f
)
{
using
namespace
caf
::
literals
;
auto
nl
=
"
\r\n
"
_sv
;
for
(;;)
{
if
(
auto
pos
=
lines
.
find
(
nl
);
pos
!=
string_view
::
npos
)
{
auto
line
=
string_view
{
lines
.
data
(),
pos
};
if
(
!
line
.
empty
())
f
(
string_view
{
lines
.
data
(),
pos
});
lines
.
remove_prefix
(
pos
+
2
);
}
else
{
return
;
}
}
}
// Splits `str` at the first occurrence of `sep` into the head and the
// remainder (excluding the separator).
std
::
pair
<
string_view
,
string_view
>
split
(
string_view
str
,
string_view
sep
)
{
auto
i
=
std
::
search
(
str
.
begin
(),
str
.
end
(),
sep
.
begin
(),
sep
.
end
());
if
(
i
!=
str
.
end
())
return
{{
str
.
begin
(),
i
},
{
i
+
sep
.
size
(),
str
.
end
()}};
return
{{
str
},
{}};
}
// Convenience function for splitting twice.
std
::
tuple
<
string_view
,
string_view
,
string_view
>
split2
(
string_view
str
,
string_view
sep
)
{
auto
[
first
,
r1
]
=
split
(
str
,
sep
);
auto
[
second
,
third
]
=
split
(
r1
,
sep
);
return
{
first
,
second
,
third
};
}
void
trim
(
string_view
&
str
)
{
auto
non_whitespace
=
[](
char
c
)
{
return
!
isspace
(
c
);
};
if
(
std
::
any_of
(
str
.
begin
(),
str
.
end
(),
non_whitespace
))
{
while
(
str
.
front
()
==
' '
)
str
.
remove_prefix
(
1
);
while
(
str
.
back
()
==
' '
)
str
.
remove_suffix
(
1
);
}
else
{
str
=
string_view
{};
}
}
bool
lowercase_equal
(
string_view
x
,
string_view
y
)
{
if
(
x
.
size
()
!=
y
.
size
())
{
return
false
;
}
else
{
for
(
size_t
index
=
0
;
index
<
x
.
size
();
++
index
)
if
(
tolower
(
x
[
index
])
!=
tolower
(
y
[
index
]))
return
false
;
return
true
;
}
}
struct
response_checker
{
string_view
ws_key
;
bool
has_status_101
=
false
;
bool
has_upgrade_field
=
false
;
bool
has_connection_field
=
false
;
bool
has_ws_accept_field
=
false
;
response_checker
(
string_view
key
)
:
ws_key
(
key
)
{
// nop
}
bool
ok
()
const
noexcept
{
return
has_status_101
&&
has_upgrade_field
&&
has_connection_field
&&
has_ws_accept_field
;
}
void
operator
()(
string_view
line
)
noexcept
{
if
(
starts_with
(
line
,
"HTTP/1"
))
{
string_view
code
;
std
::
tie
(
std
::
ignore
,
code
,
std
::
ignore
)
=
split2
(
line
,
" "
);
has_status_101
=
code
==
"101"
;
}
else
{
auto
[
field
,
value
]
=
split
(
line
,
":"
);
trim
(
field
);
trim
(
value
);
if
(
field
==
"Upgrade"
)
has_upgrade_field
=
lowercase_equal
(
value
,
"websocket"
);
else
if
(
field
==
"Connection"
)
has_connection_field
=
lowercase_equal
(
value
,
"upgrade"
);
else
if
(
field
==
"Sec-WebSocket-Accept"
)
has_ws_accept_field
=
value
==
ws_key
;
}
}
};
}
// namespace
bool
handshake
::
is_valid_http_1_response
(
string_view
http_response
)
const
{
auto
seed
=
detail
::
base64
::
encode
(
key_
);
seed
+=
"258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
;
auto
response_key_sha
=
hash
::
sha1
::
compute
(
seed
);
auto
response_key
=
detail
::
base64
::
encode
(
response_key_sha
);
response_checker
checker
{
response_key
};
for_each_http_line
(
http_response
,
checker
);
return
checker
.
ok
();
}
std
::
pair
<
string_view
,
byte_span
>
handshake
::
split_http_1_header
(
byte_span
bytes
)
{
std
::
array
<
byte
,
4
>
end_of_header
{{
byte
{
'\r'
},
byte
{
'\n'
},
byte
{
'\r'
},
byte
{
'\n'
},
}};
if
(
auto
i
=
std
::
search
(
bytes
.
begin
(),
bytes
.
end
(),
end_of_header
.
begin
(),
end_of_header
.
end
());
i
==
bytes
.
end
())
{
return
{
string_view
{},
bytes
};
}
else
{
auto
offset
=
static_cast
<
size_t
>
(
std
::
distance
(
bytes
.
begin
(),
i
));
offset
+=
end_of_header
.
size
();
return
{
string_view
{
reinterpret_cast
<
const
char
*>
(
bytes
.
begin
()),
offset
},
bytes
.
subspan
(
offset
)};
}
}
// -- utility ------------------------------------------------------------------
string_view
handshake
::
lookup
(
string_view
field_name
)
const
noexcept
{
if
(
auto
i
=
fields_
.
find
(
field_name
);
i
!=
fields_
.
end
())
return
i
->
second
;
else
return
string_view
{};
}
}
// namespace caf::net::web_socket
libcaf_net/test/net-test.hpp
View file @
4d03a100
...
...
@@ -8,7 +8,7 @@
#include "caf/span.hpp"
#include "caf/string_view.hpp"
#include "caf/tag/stream_oriented.hpp"
#include "caf/test/dsl.hpp"
#include "caf/test/
bdd_
dsl.hpp"
template
<
class
UpperLayer
>
class
mock_stream_transport
{
...
...
@@ -17,6 +17,14 @@ public:
using
output_tag
=
caf
::
tag
::
stream_oriented
;
// -- constructors, destructors, and assignment operators --------------------
template
<
class
...
Ts
>
explicit
mock_stream_transport
(
Ts
&&
...
xs
)
:
upper_layer
(
std
::
forward
<
Ts
>
(
xs
)...)
{
// nop
}
// -- interface for the upper layer ------------------------------------------
void
begin_output
()
{
...
...
libcaf_net/test/net/web_socket/client.cpp
0 → 100644
View file @
4d03a100
// 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.
#define CAF_SUITE net.web_socket.client
#include "caf/net/web_socket/client.hpp"
#include "net-test.hpp"
using
namespace
caf
;
using
namespace
caf
::
literals
;
using
namespace
std
::
literals
::
string_literals
;
namespace
{
using
svec
=
std
::
vector
<
std
::
string
>
;
struct
app_t
{
std
::
string
text_input
;
std
::
vector
<
byte
>
binary_input
;
settings
cfg
;
template
<
class
LowerLayerPtr
>
error
init
(
net
::
socket_manager
*
,
LowerLayerPtr
,
const
settings
&
init_cfg
)
{
cfg
=
init_cfg
;
return
none
;
}
template
<
class
LowerLayerPtr
>
bool
prepare_send
(
LowerLayerPtr
)
{
return
true
;
}
template
<
class
LowerLayerPtr
>
bool
done_sending
(
LowerLayerPtr
)
{
return
true
;
}
template
<
class
LowerLayerPtr
>
void
abort
(
LowerLayerPtr
,
const
error
&
reason
)
{
CAF_FAIL
(
"app::abort called: "
<<
reason
);
}
template
<
class
LowerLayerPtr
>
ptrdiff_t
consume_text
(
LowerLayerPtr
,
string_view
text
)
{
text_input
.
insert
(
text_input
.
end
(),
text
.
begin
(),
text
.
end
());
return
static_cast
<
ptrdiff_t
>
(
text
.
size
());
}
template
<
class
LowerLayerPtr
>
ptrdiff_t
consume_binary
(
LowerLayerPtr
,
byte_span
bytes
)
{
binary_input
.
insert
(
binary_input
.
end
(),
bytes
.
begin
(),
bytes
.
end
());
return
static_cast
<
ptrdiff_t
>
(
bytes
.
size
());
}
};
constexpr
auto
key
=
"the sample nonce"
_sv
;
constexpr
auto
http_request
=
"GET /chat HTTP/1.1
\r\n
"
"Host: server.example.com
\r\n
"
"Upgrade: websocket
\r\n
"
"Connection: Upgrade
\r\n
"
"Sec-WebSocket-Version: 13
\r\n
"
"Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==
\r\n
"
"Origin: http://example.com
\r\n
"
"Sec-WebSocket-Protocol: chat, superchat
\r\n
"
"
\r\n
"
_sv
;
constexpr
auto
http_response
=
"HTTP/1.1 101 Switching Protocols
\r\n
"
"Upgrade: websocket
\r\n
"
"Connection: Upgrade
\r\n
"
"Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=
\r\n
"
"
\r\n
"
_sv
;
auto
key_to_bytes
()
{
net
::
web_socket
::
handshake
::
key_type
bytes
;
auto
in
=
as_bytes
(
make_span
(
key
));
std
::
copy
(
in
.
begin
(),
in
.
end
(),
bytes
.
begin
());
return
bytes
;
}
auto
make_handshake
()
{
net
::
web_socket
::
handshake
result
;
result
.
endpoint
(
"/chat"
);
result
.
host
(
"server.example.com"
);
result
.
key
(
key_to_bytes
());
result
.
origin
(
"http://example.com"
);
result
.
protocols
(
"chat, superchat"
);
return
result
;
}
using
mock_client_type
=
mock_stream_transport
<
net
::
web_socket
::
client
<
app_t
>>
;
}
// namespace
BEGIN_FIXTURE_SCOPE
(
host_fixture
)
SCENARIO
(
"the client performs the WebSocket handshake on startup"
)
{
GIVEN
(
"valid WebSocket handshake data"
)
{
WHEN
(
"starting a WebSocket client"
){
mock_client_type
client
{
make_handshake
()};
THEN
(
"the client sends its HTTP request when initializing it"
)
{
CHECK_EQ
(
client
.
init
(),
error
{});
CHECK_EQ
(
client
.
output_as_str
(),
http_request
);
}
AND
(
"the client waits for the server handshake and validates it"
)
{
client
.
push
(
http_response
);
CHECK_EQ
(
client
.
handle_input
(),
static_cast
<
ptrdiff_t
>
(
http_response
.
size
()));
CHECK
(
client
.
upper_layer
.
handshake_complete
());
}
}
}
}
END_FIXTURE_SCOPE
()
libcaf_net/test/net/web_socket/handshake.cpp
0 → 100644
View file @
4d03a100
// 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.
#define CAF_SUITE net.web_socket.handshake
#include "caf/net/web_socket/handshake.hpp"
#include "net-test.hpp"
#include <algorithm>
using
namespace
caf
::
literals
;
using
namespace
caf
::
net
;
using
namespace
caf
;
namespace
{
constexpr
auto
key
=
"the sample nonce"
_sv
;
constexpr
auto
http_request
=
"GET /chat HTTP/1.1
\r\n
"
"Host: server.example.com
\r\n
"
"Upgrade: websocket
\r\n
"
"Connection: Upgrade
\r\n
"
"Sec-WebSocket-Version: 13
\r\n
"
"Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==
\r\n
"
"Origin: http://example.com
\r\n
"
"Sec-WebSocket-Protocol: chat, superchat
\r\n
"
"
\r\n
"
_sv
;
constexpr
auto
http_response
=
"HTTP/1.1 101 Switching Protocols
\r\n
"
"Upgrade: websocket
\r\n
"
"Connection: Upgrade
\r\n
"
"Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=
\r\n
"
"
\r\n
"
_sv
;
struct
fixture
{
byte_buffer
&
buf
()
{
return
bytes
;
}
string_view
str
()
{
return
{
reinterpret_cast
<
char
*>
(
bytes
.
data
()),
bytes
.
size
()};
}
auto
key_to_bytes
()
{
web_socket
::
handshake
::
key_type
bytes
;
auto
in
=
as_bytes
(
make_span
(
key
));
std
::
copy
(
in
.
begin
(),
in
.
end
(),
bytes
.
begin
());
return
bytes
;
}
byte_buffer
bytes
;
};
}
// namespace
BEGIN_FIXTURE_SCOPE
(
fixture
)
SCENARIO
(
"handshake generates HTTP GET requests according to RFC 6455"
)
{
GIVEN
(
"a request header object with endpoint, origin and protocol"
)
{
web_socket
::
handshake
uut
;
uut
.
endpoint
(
"/chat"
);
uut
.
host
(
"server.example.com"
);
uut
.
key
(
key_to_bytes
());
uut
.
origin
(
"http://example.com"
);
uut
.
protocols
(
"chat, superchat"
);
WHEN
(
"when generating the HTTP handshake"
)
{
THEN
(
"CAF creates output according to RFC 6455 and omits empty fields"
)
{
uut
.
write_http_1_request
(
buf
());
CHECK_EQ
(
str
(),
http_request
);
}
}
}
}
SCENARIO
(
"handshake objects validate HTTP response headers"
)
{
GIVEN
(
"a request header object with predefined key"
)
{
web_socket
::
handshake
uut
;
uut
.
endpoint
(
"/chat"
);
uut
.
key
(
key_to_bytes
());
WHEN
(
"presenting a HTTP response with proper Sec-WebSocket-Accept"
)
{
THEN
(
"the object recognizes the response as valid"
)
{
CHECK
(
uut
.
is_valid_http_1_response
(
http_response
));
CHECK
(
!
uut
.
is_valid_http_1_response
(
"HTTP/1.1 101 Bogus
\r\n
"
));
}
}
}
}
END_FIXTURE_SCOPE
()
libcaf_net/test/net/web_socket
_
server.cpp
→
libcaf_net/test/net/web_socket
/
server.cpp
View file @
4d03a100
...
...
@@ -2,9 +2,9 @@
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE net.web_socket
_
server
#define CAF_SUITE net.web_socket
.
server
#include "caf/net/web_socket
_
server.hpp"
#include "caf/net/web_socket
/
server.hpp"
#include "net-test.hpp"
...
...
@@ -103,9 +103,9 @@ struct fixture : host_fixture {
push
(
detail
::
rfc6455
::
text_frame
,
as_bytes
(
make_span
(
str
)));
}
mock_stream_transport
<
net
::
web_socket
_
server
<
app_t
>>
transport
;
mock_stream_transport
<
net
::
web_socket
::
server
<
app_t
>>
transport
;
net
::
web_socket
_
server
<
app_t
>*
ws
;
net
::
web_socket
::
server
<
app_t
>*
ws
;
app_t
*
app
;
...
...
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