Commit 46df74f8 authored by Dominik Charousset's avatar Dominik Charousset

Add simple HTTP broker example

parent 01f2c6d0
......@@ -35,7 +35,8 @@ add(typed_calculator message_passing)
add(distributed_calculator remote_actors)
add(group_server remote_actors)
add(group_chat remote_actors)
add(simple_broker remote_actors)
add(simple_broker brokers)
add(simple_http_broker brokers)
if(NOT CAF_NO_PROTOBUF_EXAMPLES)
find_package(Protobuf)
......@@ -44,7 +45,7 @@ if(NOT CAF_NO_PROTOBUF_EXAMPLES)
include_directories(${PROTOBUF_INCLUDE_DIR})
# add binary dir as include path as generated headers will be located there
include_directories(${CMAKE_CURRENT_BINARY_DIR})
add_executable(protobuf_broker remote_actors/protobuf_broker.cpp ${ProtoSources})
add_executable(protobuf_broker brokers/protobuf_broker.cpp ${ProtoSources})
target_link_libraries(protobuf_broker ${CMAKE_DL_LIBS} ${LIBCAF_LIBRARIES} ${PTHREAD_LIBRARIES} ${PROTOBUF_LIBRARIES})
add_dependencies(protobuf_broker all_examples)
endif(PROTOBUF_FOUND AND PROTOBUF_PROTOC_EXECUTABLE)
......
#include <iostream>
#include "caf/all.hpp"
#include "caf/io/all.hpp"
using std::cout;
using std::cerr;
using std::endl;
using namespace caf;
using namespace caf::io;
constexpr const char http_ok[] = R"__(HTTP/1.1 200 OK
Content-Type: text/plain
Connection: keep-alive
Transfer-Encoding: chunked
d
Hi there! :)
0
)__";
template <size_t Size>
constexpr size_t cstr_size(const char (&)[Size]) {
return Size;
}
actor_ostream out(broker* self, const char* role) {
return aout(self) << "[" << role << ":" << self->id() << "] ";
}
actor_ostream wout(broker* self) {
return out(self, "worker");
}
actor_ostream sout(broker* self) {
return out(self, "server");
}
behavior connection_worker(broker* self, connection_handle hdl) {
self->configure_read(hdl, receive_policy::at_most(1024));
return {
[=](const new_data_msg& msg) {
wout(self) << "received data, send http_ok" << endl;
self->write(msg.handle, cstr_size(http_ok), http_ok);
self->quit();
},
[=](const connection_closed_msg&) {
wout(self) << "connection closed by remote host, quit" << endl;
self->quit();
}
};
}
behavior server(broker* self) {
sout(self) << "running" << endl;
return {
[=](const new_connection_msg& ncm) {
auto worker = self->fork(connection_worker, ncm.handle);
self->monitor(worker);
self->link_to(worker);
sout(self) << "forked connection, worker ID: " << worker.id() << endl;
},
[=](const down_msg& dm) {
sout(self) << "worker with ID " << dm.source.id() << " is done" << endl;
},
others() >> [=] {
sout(self) << "unexpected: " << to_string(self->last_dequeued()) << endl;
}
};
}
optional<uint16_t> as_u16(const std::string& str) {
return static_cast<uint16_t>(stoul(str));
}
int main(int argc, const char **argv) {
message_builder{argv + 1, argv + argc}.apply({
on("-p", as_u16) >> [&](uint16_t port) {
cout << "*** run in server mode listen on: " << port << endl;
cout << "*** to quit the program, simply press <enter>" << endl;
auto sever_actor = spawn_io_server(server, port);
// wait for any input
std::string dummy;
std::getline(std::cin, dummy);
// kill server
anon_send_exit(sever_actor, exit_reason::user_shutdown);
},
others() >> [] {
cerr << "use with '-p PORT' as server on port" << endl;
}
});
await_all_actors_done();
shutdown();
}
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment