Begin websocket refactor (75a0ca14) · Commits · olcf / container-builder

Builder/include/Builder.h

+9 −7

Original line number	Diff line number	Diff line
		@@ -7,6 +7,7 @@
		#include <boost/beast/websocket.hpp>
		#include <boost/asio/ip/tcp.hpp>
		#include <boost/asio/spawn.hpp>
		#include <boost/asio/buffer.hpp>
		#include <boost/process.hpp>
		#include <boost/regex.hpp>
		#include <boost/filesystem.hpp>
		@@ -74,21 +75,22 @@ public:
		logger::write("launched build process: " + build_command);

		// stream from the async pipe process output to the socket
		asio::streambuf buffer;
		std::array<char, 4096> buffer;
		std::size_t read_size = 0;
		boost::system::error_code stream_error;
		bool fin = false;
		do {
		// Read from the pipe into a buffer
		std_pipe.async_read_some(buffer, yield[stream_error]);
		auto read_bytes = std_pipe.async_read_some(buffer, yield[stream_error]);
		if (stream_error != boost::system::errc::success && stream_error != boost::asio::error::eof) {
		throw std::runtime_error("reading process pipe failed: " + stream_error.message());
		}
		if(read_size == 0 \|\| stream_error) {
		// Wrap it up if we've hit EOF
		if(stream_error == boost::asio::error::eof) {
		fin = true;
		}
		// Write the buffer to our socket
		client.async_write_some_streambuf(fin, buffer, yield, error);
		client.async_write_some(fin, asio::buffer(buffer, read_bytes), yield, error);
		if (error) {
		throw std::runtime_error("sending process pipe failed: " + error.message());
		}
		@@ -102,9 +104,9 @@ public:
		// Send the container to the client
		if (build_code == 0) {
		logger::write("Build complete, sending container");
		client.async_send_file("container.img");
		if (client.error) {
		throw std::runtime_error("Sending file to client failed: " + client.error.message());
		client.async_write_file("container.img", yield, error);
		if (error) {
		throw std::runtime_error("Sending file to client failed: " + error.message());
		}
		} else {
		throw std::runtime_error("Build failed, not sending container");

Common/include/Messenger.h

+2 −1

Original line number	Diff line number	Diff line
		@@ -68,7 +68,8 @@ public:
		asio::yield_context yield,
		boost::system::error_code& error);

		void async_write_streambuf(asio::streambuf &message_body,
		void async_write_some_streambuf(bool fin,
		asio::streambuf &message,
		asio::yield_context yield,
		boost::system::error_code& error);

Common/src/Messenger.cpp

+30 −24

Original line number	Diff line number	Diff line
		@@ -39,19 +39,21 @@ void Messenger::async_read_file(boost::filesystem::path file_path,
		file.open(file_path.string(), std::fstream::out \| std::fstream::binary \| std::fstream::trunc);
		if (!file) {
		error = boost::system::errc::make_error_code(boost::system::errc::io_error);
		logger::write("Error opening file: " + file_path.string() + " " + std::strerror(errno), logger::severity_level::error);
		logger::write("Error opening file: " + file_path.string() + " " + std::strerror(errno),
		logger::severity_level::error);
		return;
		}

		const auto chunk_size = 4096;
		char buffer[chunk_size];
		std::array<char, chunk_size> buffer;
		boost::crc_32_type csc_result;

		// Read file in chunks
		do {
		auto bytes_read = stream.async_read_some(buffer, chunk_size, yield[error]);
		if (error != beast::errc::success && error != asio::error::eof) {
		logger::write("Error reading file: " + file_path.string() + " " + std::strerror(errno), logger::severity_level::error);
		logger::write("Error reading file: " + file_path.string() + " " + std::strerror(errno),
		logger::severity_level::error);
		file.close();
		return;
		}
		@@ -62,7 +64,8 @@ void Messenger::async_read_file(boost::filesystem::path file_path,
		file.close();
		if (!file) {
		error = boost::system::errc::make_error_code(boost::system::errc::io_error);
		logger::write("Error closing file: " + file_path.string() + " " + std::strerror(errno), logger::severity_level::error);
		logger::write("Error closing file: " + file_path.string() + " " + std::strerror(errno),
		logger::severity_level::error);
		}

		// Read remote file checksum and verify
		@@ -81,11 +84,12 @@ void Messenger::async_read_file(boost::filesystem::path file_path,
		}

		// Send a streambuf message asynchronously
		void Messenger::async_write_steambuf(asio::streambuf &message,
		void Messenger::async_write_some_steambuf(bool fin,
		asio::streambuf &message,
		asio::yield_context yield,
		boost::system::error_code &error) {
		// Write the message body
		stream.async_write(message, yield[error]);
		stream.async_write_some(fin, message, yield[error]);
		}

		// Send a file as a message asynchronously
		@@ -97,12 +101,13 @@ void Messenger::async_write_file(boost::filesystem::path file_path,
		// Open file and get size
		file.open(file_path.string(), std::fstream::in \| std::fstream::binary);
		if (!file) {
		logger::write("Error opening file: " + file_path.string() + " " + std::strerror(errno), logger::severity_level::error);
		logger::write("Error opening file: " + file_path.string() + " " + std::strerror(errno),
		logger::severity_level::error);
		}
		auto file_size = boost::filesystem::file_size(file_path);

		const auto chunk_size = 4096;
		char buffer[chunk_size];
		std::array<char, chunk_size> buffer;
		boost::crc_32_type csc_result;

		auto bytes_remaining = file_size;
		@@ -129,7 +134,8 @@ void Messenger::async_write_file(boost::filesystem::path file_path,
		file.close();
		if (!file) {
		error = boost::system::errc::make_error_code(boost::system::errc::io_error);
		logger::write("Error closing file: " + file_path.string() + " " + std::strerror(errno), logger::severity_level::error);
		logger::write("Error closing file: " + file_path.string() + " " + std::strerror(errno),
		logger::severity_level::error);
		}

		// After we've sent the file we send the checksum