Actually log a session from client instead of pinging

The Client class provides a nice, high-level async API for
client-server communication.

client/src/activities_view.vala

@@ -1,28 +1,10 @@
 namespace StudySystemClient {
-	private struct Activity {
-		public string subject;
-		public ActivityType type;
-	}
-
-	enum ActivityType {
-		EXERCISES,
-		READING;
-
-		public string to_string() {
-			switch (this) {
-			case EXERCISES:
-				return "Exercises";
-			case READING:
-				return "Reading";
-			default:
-				return "Invalid activity type";
-			}
-		}
-	}
-
 	public class ActivitiesView : Gtk.Box {
-		public ActivitiesView()  {
+		private Client client;
+
+		public ActivitiesView(Client client)  {
 			margin_top = margin_bottom = margin_start = margin_end = 0;
+			this.client = client;
 
 			var scrolled_window = new Gtk.ScrolledWindow();
 			scrolled_window.hscrollbar_policy = Gtk.PolicyType.NEVER;
@@ -44,15 +26,31 @@ namespace StudySystemClient {
 				{ "Cybernetics", ActivityType.READING },
 				{ "Physics", ActivityType.EXERCISES },
 			};
-			foreach (var activity in activities)
+			foreach (var activity in activities) {
-				card_container.append(new ActivityCard(activity));
+				var card = new ActivityCard(activity);
+				card.session_logged.connect(log_session);
+				card_container.append(card);
+			}
 
 			scrolled_window.set_child(card_container);
 			this.append(scrolled_window);
 		}
+
+		private async void log_session(string subject, ActivityType type,
+									   int minutes) {
+			try {
+				yield client.log_session(subject, type, minutes);
+				stderr.printf("Successfully logged session\n");
+			} catch (ClientError e) {
+				stderr.printf("Error logging session: %s\n", e.message);
+			}
+		}
 	}
 
 	private class ActivityCard : Gtk.Frame {
+		public signal void session_logged(string subject, ActivityType type,
+										  int minutes);
+
 		public ActivityCard(Activity activity) {
 			add_css_class("card");
 
@@ -85,10 +83,15 @@ namespace StudySystemClient {
 			var log_session_popover = new LogSessionPopover();
 			log_session_popover.set_parent(button);
 			button.clicked.connect(() => log_session_popover.popup());
+			log_session_popover.session_logged.connect((minutes) => {
+					session_logged(activity.subject, activity.type, minutes);
+				});
 		}
 	}
 
 	private class LogSessionPopover : Gtk.Popover {
+		public signal void session_logged(int minutes);
+
 		private const int DEFAULT_LENGTH = 30;
 
 		private Gtk.SpinButton input;
@@ -119,6 +122,7 @@ namespace StudySystemClient {
 		}
 
 		private void submit() {
+			session_logged((int)input.value);
 			reset();
 			popdown();
 		}

client/src/activity.vala

@@ -0,0 +1,23 @@
+namespace StudySystemClient {
+	public struct Activity {
+		public string subject;
+		public ActivityType type;
+		public double priority;
+	}
+
+	public enum ActivityType {
+		READING = 0,
+		EXERCISES = 1;
+
+		public string to_string() {
+			switch (this) {
+			case EXERCISES:
+				return "Exercises";
+			case READING:
+				return "Reading";
+			default:
+				return "Invalid activity type";
+			}
+		}
+	}
+}

client/src/client.vala

@@ -0,0 +1,62 @@
+namespace StudySystemClient {
+	public errordomain ClientError {
+		ERROR_RESPONSE,
+		UNEXPECTED_RESPONSE,
+	}
+
+	public class Client {
+		private Connection connection;
+
+		public Client(string cert_dir) throws Error {
+			connection = new Connection(cert_dir);
+		}
+
+		public async void ping() throws ClientError {
+			var response = yield connection.send(new Request.Ping());
+			if (response is Response.Ack) {
+				return;
+			} else if (response is Response.Error) {
+				throw new ClientError.ERROR_RESPONSE(
+					"Error response to Ping: %s",
+					response.value.to_string());
+			} else {
+				throw new ClientError.UNEXPECTED_RESPONSE(
+					"Unexpected response to Ping");
+			}
+		}
+
+		public async Array<Activity> list_activities()
+				throws ClientError {
+			var request = new Request.ListActivities();
+			var response = yield connection.send(request);
+			if (response is Response.Activities) {
+				return response.value;
+			} else if (response is Response.Error) {
+				throw new ClientError.ERROR_RESPONSE(
+					"Error response to ListActivities: %s",
+					response.value.to_string());
+			} else {
+				throw new ClientError.UNEXPECTED_RESPONSE(
+					"Unexpected response to ListActivities");
+			}
+		}
+
+		public async void log_session(string subject, ActivityType type,
+									  int minutes) throws ClientError {
+			var timestamp = new DateTime.now_utc().to_unix();
+			var request
+				= new Request.LogSession(subject, type, timestamp, minutes);
+			var response = yield connection.send(request);
+			if (response is Response.Ack) {
+				return;
+			} else if (response is Response.Error) {
+				throw new ClientError.ERROR_RESPONSE(
+					"Error response to LogSession: %s",
+					response.value.to_string());
+			} else {
+				throw new ClientError.UNEXPECTED_RESPONSE(
+					"Unexpected response to LogSession");
+			}
+		}
+	}
+}

client/src/connection.vala

@@ -1,8 +1,7 @@
 namespace StudySystemClient {
 	public class Connection {
-		public signal void received(uint8[] msg);
-
 		private SessionManager session_manager;
+		private TransactionManager transaction_manager;
 		private Worker worker;
 
 		public Connection(string cert_dir) throws Error {
@@ -10,19 +9,78 @@ namespace StudySystemClient {
 			var session_factory
 				= new SessionFactory(loopback, 12888, cert_dir);
 			session_manager = new SessionManager(
-				session_factory, (msg) => {
+				session_factory, (msg) => receive(msg));
-					var msg_copy = new uint8[msg.length];
+			transaction_manager = new TransactionManager();
-					Memory.copy(msg_copy, msg, msg.length);
-					Idle.add(() => {
-							received(msg_copy);
-							return false;
-						}, GLib.Priority.DEFAULT_IDLE);
-				});
 			worker = new Worker(session_manager);
 		}
 
-		public void send(owned uint8[] msg) {
+		public async Response.Body? send(Request.Body body) {
-			session_manager.send(msg);
+			var transaction_id = transaction_manager.register(send.callback);
+			var request = new Request.Request(transaction_id, body);
+			session_manager.send(request.encode());
+			yield;
+			return transaction_manager.get_result(transaction_id);
+		}
+
+		private void receive(owned uint8[] msg) {
+			Response.Response response;
+			try {
+				response = new Response.Response.from_bytes(msg);
+			} catch (Response.DecodeError e) {
+				stderr.printf("Invalid response from server: %s", e.message);
+				return;
+			}
+			Idle.add(() => {
+					transaction_manager.resolve(response);
+					return false;
+				}, GLib.Priority.DEFAULT_IDLE);
+		}
+	}
+
+	private class Continuation {
+		private SourceFunc callback;
+
+		public Continuation(owned SourceFunc callback) {
+			this.callback = (owned) callback;
+		}
+
+		public void resume() {
+			callback();
+		}
+	}
+
+	private class TransactionManager {
+		private uint16 next_transaction_id;
+		private HashTable<uint16, Continuation> pending;
+		private HashTable<uint16, Response.Body> results;
+
+		public TransactionManager() {
+			next_transaction_id = 0;
+			pending = new HashTable<uint16, Continuation>(null, null);
+			results = new HashTable<uint16, Response.Body>(null, null);
+		}
+
+		public uint16 register(owned SourceFunc callback) {
+			var transaction_id = next_transaction_id++;
+			pending.insert(transaction_id,
+						   new Continuation((owned) callback));
+			return transaction_id;
+		}
+
+		public void resolve(Response.Response response) {
+			var transaction_id = (uint16)response.transaction_id;
+			var continuation = pending.lookup(transaction_id);
+			if (continuation == null) {
+				stderr.printf("Response for non-pending transaction %d\n",
+							  transaction_id);
+				return;
+			}
+			results.insert(transaction_id, response.body);
+			continuation.resume();
+		}
+
+		public Response.Body? get_result(uint16 transaction_id) {
+			return results.lookup(transaction_id);
 		}
 	}
 
@@ -50,172 +108,5 @@ namespace StudySystemClient {
 				Thread.usleep(1000 * TASK_PERIOD_MS);
 			}
 		}
-
-	}
-
-	private class SessionManager {
-		public delegate void ReceiveCallback(uint8[] msg);
-
-		private const uint INIT_RECONNECT_WAIT_MS = 500;
-		private const uint MAX_RECONNECT_WAIT_MS = 60000;
-		private const double RECONNECT_BACKOFF = 1.6;
-
-		private SessionFactory session_factory;
-		private ReceiveCallback receive_callback;
-		private Session? session;
-		private AsyncQueue<OutgoingMessage> queue;
-		private uint reconnect_wait_ms;
-
-		public SessionManager(SessionFactory session_factory,
-							  owned ReceiveCallback receive_callback) {
-			this.session_factory = session_factory;
-			this.receive_callback = (owned) receive_callback;
-			this.session = null;
-			queue = new AsyncQueue<OutgoingMessage>();
-			reconnect_wait_ms = INIT_RECONNECT_WAIT_MS;
-		}
-
-		public void send(uint8[] msg) {
-			queue.push(new OutgoingMessage(msg));
-		}
-
-		public void task() {
-			if (session != null) {
-				var failed_msg = session.task(queue);
-				if (failed_msg != null)
-					handle_failed_msg(failed_msg);
-			} else {
-				try_start_session();
-			}
-		}
-
-		private void handle_failed_msg(OutgoingMessage msg) {
-			msg.has_failed();
-			if (msg.should_retry())
-				queue.push(msg);
-			session = null;
-		}
-
-		private void try_start_session() {
-			try {
-				session = session_factory.start_session();
-				session.received.connect(
-					(msg) => receive_callback(msg));
-				reconnect_wait_ms = INIT_RECONNECT_WAIT_MS;
-			} catch (Error _) {
-				Thread.usleep(1000 * reconnect_wait_ms);
-				update_reconnect_wait();
-			}
-		}
-
-		private void update_reconnect_wait() {
-			var new_wait = RECONNECT_BACKOFF * reconnect_wait_ms;
-			if (new_wait < MAX_RECONNECT_WAIT_MS)
-				reconnect_wait_ms = (uint)new_wait;
-			else
-				reconnect_wait_ms = MAX_RECONNECT_WAIT_MS;
-		}
-	}
-
-	private class SessionFactory {
-		private const string CA_FILENAME = "/ca.pem";
-		private const string CERT_FILENAME = "/client.pem";
-		private const uint TIMEOUT_S = 1;
-
-		private InetSocketAddress host;
-		private TlsCertificate cert;
-		private TlsDatabase ca_db;
-
-		public SessionFactory(InetAddress host_addr, uint16 host_port,
-							  string cert_dir) throws Error {
-			host = new InetSocketAddress(host_addr, host_port);
-			var cert_path = cert_dir + CERT_FILENAME;
-			cert = new TlsCertificate.from_file(cert_path);
-			var ca_path = cert_dir + CA_FILENAME;
-			var db_type = TlsBackend.get_default().get_file_database_type();
-			ca_db = Object.new(db_type, "anchors", ca_path) as TlsDatabase;
-		}
-
-		public Session start_session() throws Error {
-			var plain_client = new SocketClient();
-			plain_client.set_timeout(TIMEOUT_S);
-			var plain_connection = plain_client.connect(host);
-			var connection = TlsClientConnection.new(plain_connection, host);
-			connection.set_database(ca_db);
-			connection.set_certificate(cert);
-			connection.handshake();
-			return new Session(connection);
-		}
-	}
-
-	private class Session {
-		public signal void received(uint8[] msg);
-
-		private const uint MAX_BATCH_SIZE = 10;
-		private const uint MAX_MSG_LEN = 1024;
-
-		private TlsClientConnection connection;
-
-		public Session(TlsClientConnection connection) {
-			this.connection = connection;
-		}
-
-		public OutgoingMessage? task(AsyncQueue<OutgoingMessage> queue) {
-			for (int i = 0; i < MAX_BATCH_SIZE; ++i) {
-				if (queue.length() == 0)
-					break;
-				var msg = queue.pop();
-				var success = true;
-				success &= send(msg);
-				success &= receive();
-				if (!success)
-					return msg;
-			}
-			return null;
-		}
-
-		private bool send(OutgoingMessage msg) {
-			try {
-				size_t written;
-				connection.output_stream.write_all(msg.content, out written);
-				return true;
-			} catch (IOError _) {
-				return false;
-			}
-		}
-
-		private bool receive() {
-			try {
-				var buffer = new uint8[MAX_MSG_LEN];
-				var len = connection.input_stream.read(buffer);
-				if (len <= 0)
-					return false;
-				received(buffer[0:len]);
-				return true;
-			} catch (IOError _) {
-				return false;
-			}
-		}
-	}
-
-	private class OutgoingMessage {
-		public uint8[] content { get; private set; }
-
-		private const uint MAX_FAIL_COUNT = 4;
-
-		private uint fail_count;
-
-		public OutgoingMessage(owned uint8[] content) {
-			this.content = (owned)content;
-			fail_count = 0;
-		}
-
-		public void has_failed() {
-			++fail_count;
-		}
-
-		public bool should_retry() {
-			return fail_count < MAX_FAIL_COUNT;
-		}
 	}
 }

client/src/der.vala

@@ -71,8 +71,8 @@ namespace StudySystemClient.Der {
 	}
 
 	public abstract class Datum {
-		internal uint8 type;
+		protected uint8 type;
-		internal uint8[] content;
+		protected uint8[] content;
 
 		public uint8[] encode() {
 			var buffer = new ByteArray();

client/src/main.vala

@@ -14,8 +14,8 @@ namespace StudySystemClient {
 					css_provider,
 					Gtk.STYLE_PROVIDER_PRIORITY_APPLICATION);
 
-				var connection = new Connection(Config.CERT_DIR);
+				var client = new Client(Config.CERT_DIR);
-				var main_window = new MainWindow(this, connection);
+				var main_window = new MainWindow(this, client);
 				main_window.present();
 			} catch (Error e) {
 				stderr.printf("Failed to initialize: %s\n", e.message);

client/src/main_window.vala

@@ -1,22 +1,18 @@
 namespace StudySystemClient {
 	public class MainWindow : Gtk.ApplicationWindow {
-		private Connection connection;
+		public MainWindow(Gtk.Application app, Client client) {
-
-		public MainWindow(Gtk.Application app, Connection connection) {
 			Object(application: app);
 
 			default_width = 360;
 			default_height = 580;
 
-			this.connection = connection;
-
 			var header_bar = new Gtk.HeaderBar();
 			var title = new Gtk.Label("Study System Client");
 			title.add_css_class("title");
 			header_bar.title_widget = title;
 			set_titlebar(header_bar);
 
-			var activities_view = new ActivitiesView();
+			var activities_view = new ActivitiesView(client);
 			set_child(activities_view);
 		}
 	}

client/src/meson.build

@@ -10,10 +10,15 @@ configure_file(
 lib = library(
   'study-system-client',
   sources: files(
+    'activity.vala',
     'activities_view.vala',
+    'client.vala',
     'connection.vala',
     'der.vala',
     'main_window.vala',
+    'request.vala',
+    'response.vala',
+    'session_manager.vala',
   ) + resources,
   dependencies: [gtk_dep],
   vala_vapi: 'study-system-client.vapi',

client/src/request.vala

@@ -0,0 +1,54 @@
+namespace StudySystemClient.Request {
+	public class Request {
+		private Der.Datum datum;
+
+		public Request(uint16 transaction_id, Body body) {
+			datum = new Der.Sequence(
+				{ new Der.Integer(transaction_id), body.datum });
+		}
+
+		public uint8[] encode() {
+			return datum.encode();
+		}
+	}
+
+	public abstract class Body {
+		protected enum Tag {
+			PING = 0,
+			LIST_ACTIVITIES = 1,
+			LOG_SESSION = 2,
+		}
+
+		internal Der.Datum datum;
+
+		protected Body(Tag tag, Der.Datum datum) {
+			this.datum = new Der.Choice(tag, datum);
+		}
+	}
+
+	public class Ping : Body {
+		public Ping() {
+			base(Tag.PING, new Der.Null());
+		}
+	}
+
+	public class ListActivities : Body {
+		public ListActivities() {
+			base(Tag.LIST_ACTIVITIES, new Der.Null());
+		}
+	}
+
+	public class LogSession : Body {
+		public LogSession(string subject, ActivityType type,
+						  int64 timestamp, int minutes)
+		{
+			var fields = new Der.Datum[] {
+				new Der.Utf8String(subject),
+				new Der.Enumerated((int)type),
+				new Der.Integer(timestamp),
+				new Der.Integer(minutes),
+			};
+			base(Tag.LOG_SESSION, new Der.Sequence(fields));
+		}
+	}
+}

client/src/response.vala

@@ -0,0 +1,175 @@
+namespace StudySystemClient.Response {
+	public class Response {
+		public int transaction_id { get; private set; }
+		public Body body { get; private set; }
+
+		public Response.from_bytes(uint8[] bytes) throws DecodeError {
+			Der.Sequence sequence;
+			try {
+				sequence = Der.decode(bytes) as Der.Sequence;
+			} catch (Der.DecodeError e) {
+				throw new DecodeError.INVALID_RESPONSE(
+					"Response was not valid DER: " + e.message);
+			}
+			if (sequence == null) {
+				throw new DecodeError.INVALID_RESPONSE(
+					"Response was not a SEQUENCE");
+			}
+			if (sequence.value.length < 2) {
+				throw new DecodeError.INVALID_RESPONSE(
+					"Too few fields in Response: %u (expected 2)",
+					sequence.value.length);
+			}
+			var id_datum = sequence.value[0] as Der.Integer;
+			if (id_datum == null) {
+				throw new DecodeError.INVALID_RESPONSE(
+					"Response transactionId was not an INTEGER");
+			}
+			transaction_id = (int)id_datum.value;
+			body = Body.from_datum(sequence.value[1]);
+		}
+	}
+
+	public errordomain DecodeError {
+		INVALID_BODY,
+		INVALID_RESPONSE,
+		NOT_IMPLEMENTED,
+	}
+
+	public abstract class Body {
+		protected enum Tag {
+			ERROR = 0,
+			ACK = 1,
+			PRIORITIZED_ACTIVITIES = 2,
+		}
+
+		internal static Body from_datum(Der.Datum datum) throws DecodeError {
+			var choice = datum as Der.Choice;
+			if (choice == null) {
+				throw new DecodeError.INVALID_BODY(
+					"ResponseBody was not a CHOICE");
+			}
+			switch (choice.id) {
+			case Tag.ERROR:
+				return new Error.from_datum(choice.value);
+			case Tag.ACK:
+				return new Ack.from_datum(choice.value);
+			case Tag.PRIORITIZED_ACTIVITIES:
+				throw new DecodeError.NOT_IMPLEMENTED(
+					"PrioritizedActivities not yet implemented");
+			default:
+				throw new DecodeError.INVALID_BODY(
+					"Invalid ResponseBody tag");
+			}
+		}
+	}
+
+	public class Error : Body {
+		public enum Value {
+			INVALID_REQUEST = 0,
+			INVALID_ARGUMENTS = 1,
+			SERVER_ERROR = 2;
+
+			public string to_string() {
+				switch (this) {
+				case INVALID_REQUEST:
+					return "Invalid request";
+				case INVALID_ARGUMENTS:
+					return "Invalid arguments";
+				case SERVER_ERROR:
+					return "Server error";
+				default:
+					return "Unknown error";
+				}
+			}
+		}
+
+		public Value value { get; private set; }
+
+		internal Error.from_datum(Der.Datum datum) throws DecodeError {
+			var enumerated = datum as Der.Enumerated;
+			if (enumerated == null) {
+				throw new DecodeError.INVALID_BODY(
+					"Error was not an ENUMERATED");
+			}
+			if (enumerated.value < 0 || enumerated.value > 2) {
+				throw new DecodeError.INVALID_BODY(
+					"Error type was not in range 0..2");
+			}
+			value = (Value)enumerated.value;
+		}
+	}
+
+	public class Ack : Body {
+		internal Ack.from_datum(Der.Datum datum) throws DecodeError {
+			var @null = datum as Der.Null;
+			if (@null == null) {
+				throw new DecodeError.INVALID_BODY("Ack was not NULL");
+			}
+		}
+	}
+
+	public class Activities : Body {
+		public Array<Activity> value { get; private set; }
+
+		internal Activities.from_datum(Der.Datum datum) throws DecodeError {
+			value = new Array<Activity>();
+			if (datum is Der.Sequence) {
+				foreach (var activity_datum in datum.value)
+					value.append_val(activity_from_datum(activity_datum));
+			} else {
+				throw new DecodeError.INVALID_BODY(
+					"Activities was not a SEQUENCE");
+			}
+		}
+
+		private static Activity activity_from_datum(Der.Datum datum)
+			throws DecodeError {
+			if (datum is Der.Sequence) {
+				var fields = datum.value;
+				if (fields.length < 3) {
+					throw new DecodeError.INVALID_BODY(
+						"Too few fields in Activity: %u (expected 3)",
+						fields.length);
+				}
+				var subject = get_string("Activity.subject", fields[0]);
+				var activity_type
+					= get_activity_type("Activity.type", fields[1]);
+				var priority_int = get_int("Activity.priority", fields[2]);
+				var priority = (double)priority_int / 100.0;
+				return { subject, activity_type, priority };
+			} else {
+				throw new DecodeError.INVALID_BODY(
+					"Activity was not a SEQUENCE");
+			}
+		}
+
+		private static int get_int(string name, Der.Datum datum)
+			throws DecodeError {
+			if (datum is Der.Integer)
+				return (int)datum.value;
+			throw new DecodeError.INVALID_BODY(@"$name was not an INTEGER");
+		}
+
+		private static string get_string(string name, Der.Datum datum)
+			throws DecodeError {
+			if (datum is Der.Utf8String)
+				return datum.value;
+			throw new DecodeError.INVALID_BODY(
+				@"$name was not a UTF8String");
+		}
+
+		private static ActivityType get_activity_type(
+			string name, Der.Datum datum) throws DecodeError {
+			if (datum is Der.Enumerated) {
+				var value = datum.value;
+				if (0 <= value <= 1)
+					return (ActivityType)value;
+				throw new DecodeError.INVALID_BODY(
+					"Invalid value for ActivityType: %lld", value);
+			}
+			throw new DecodeError.INVALID_BODY(
+				@"$name was not an ENUMERATED");
+		}
+	}
+}

client/src/session_manager.vala

@@ -0,0 +1,166 @@
+namespace StudySystemClient {
+	public class SessionManager {
+		public delegate void ReceiveCallback(owned uint8[] msg);
+
+		private const uint INIT_RECONNECT_WAIT_MS = 500;
+		private const uint MAX_RECONNECT_WAIT_MS = 60000;
+		private const double RECONNECT_BACKOFF = 1.6;
+
+		private SessionFactory session_factory;
+		private ReceiveCallback receive_callback;
+		private Session? session;
+		private AsyncQueue<OutgoingMessage> queue;
+		private uint reconnect_wait_ms;
+
+		public SessionManager(SessionFactory session_factory,
+							  owned ReceiveCallback receive_callback) {
+			this.session_factory = session_factory;
+			this.receive_callback = (owned) receive_callback;
+			this.session = null;
+			queue = new AsyncQueue<OutgoingMessage>();
+			reconnect_wait_ms = INIT_RECONNECT_WAIT_MS;
+		}
+
+		public void send(uint8[] msg) {
+			queue.push(new OutgoingMessage(msg));
+		}
+
+		public void task() {
+			if (session != null) {
+				var failed_msg = session.task(queue);
+				if (failed_msg != null)
+					handle_failed_msg(failed_msg);
+			} else {
+				try_start_session();
+			}
+		}
+
+		private void handle_failed_msg(OutgoingMessage msg) {
+			msg.has_failed();
+			if (msg.should_retry())
+				queue.push(msg);
+			session = null;
+		}
+
+		private void try_start_session() {
+			try {
+				session = session_factory.start_session();
+				session.received.connect((msg) => receive_callback(msg));
+				reconnect_wait_ms = INIT_RECONNECT_WAIT_MS;
+			} catch (Error _) {
+				Thread.usleep(1000 * reconnect_wait_ms);
+				update_reconnect_wait();
+			}
+		}
+
+		private void update_reconnect_wait() {
+			var new_wait = RECONNECT_BACKOFF * reconnect_wait_ms;
+			if (new_wait < MAX_RECONNECT_WAIT_MS)
+				reconnect_wait_ms = (uint)new_wait;
+			else
+				reconnect_wait_ms = MAX_RECONNECT_WAIT_MS;
+		}
+	}
+
+	public class SessionFactory {
+		private const string CA_FILENAME = "/ca.pem";
+		private const string CERT_FILENAME = "/client.pem";
+		private const uint TIMEOUT_S = 5;
+
+		private InetSocketAddress host;
+		private TlsCertificate cert;
+		private TlsDatabase ca_db;
+
+		public SessionFactory(InetAddress host_addr, uint16 host_port,
+							  string cert_dir) throws Error {
+			host = new InetSocketAddress(host_addr, host_port);
+			var cert_path = cert_dir + CERT_FILENAME;
+			cert = new TlsCertificate.from_file(cert_path);
+			var ca_path = cert_dir + CA_FILENAME;
+			var db_type = TlsBackend.get_default().get_file_database_type();
+			ca_db = Object.new(db_type, "anchors", ca_path) as TlsDatabase;
+		}
+
+		internal Session start_session() throws Error {
+			var plain_client = new SocketClient();
+			plain_client.set_timeout(TIMEOUT_S);
+			var plain_connection = plain_client.connect(host);
+			var connection = TlsClientConnection.new(plain_connection, host);
+			connection.set_database(ca_db);
+			connection.set_certificate(cert);
+			connection.handshake();
+			return new Session(connection);
+		}
+	}
+
+	private class Session {
+		public signal void received(uint8[] msg);
+
+		private const uint MAX_BATCH_SIZE = 10;
+		private const uint MAX_MSG_LEN = 1024;
+
+		private TlsClientConnection connection;
+
+		public Session(TlsClientConnection connection) {
+			this.connection = connection;
+		}
+
+		public OutgoingMessage? task(AsyncQueue<OutgoingMessage> queue) {
+			for (int i = 0; i < MAX_BATCH_SIZE; ++i) {
+				if (queue.length() == 0)
+					break;
+				var msg = queue.pop();
+				var success = true;
+				success &= send(msg);
+				success &= receive();
+				if (!success)
+					return msg;
+			}
+			return null;
+		}
+
+		private bool send(OutgoingMessage msg) {
+			try {
+				size_t written;
+				connection.output_stream.write_all(msg.content, out written);
+				return true;
+			} catch (IOError _) {
+				return false;
+			}
+		}
+
+		private bool receive() {
+			try {
+				var buffer = new uint8[MAX_MSG_LEN];
+				var len = connection.input_stream.read(buffer);
+				if (len <= 0)
+					return false;
+				received(buffer[0:len]);
+				return true;
+			} catch (IOError _) {
+				return false;
+			}
+		}
+	}
+
+	private class OutgoingMessage {
+		public uint8[] content { get; private set; }
+
+		private const uint MAX_FAIL_COUNT = 4;
+
+		private uint fail_count;
+
+		public OutgoingMessage(owned uint8[] content) {
+			this.content = (owned)content;
+			fail_count = 0;
+		}
+
+		public void has_failed() {
+			++fail_count;
+		}
+
+		public bool should_retry() {
+			return fail_count < MAX_FAIL_COUNT;
+		}
+	}
+}

server/.gitignore

@@ -1,3 +1,4 @@
 _build/*
 *.beam
 src/StudySystemProtocol.erl
+include/*

server/asn1/StudySystemProtocol.asn1

@@ -1,11 +1,49 @@
 StudySystemProtocol DEFINITIONS EXPLICIT TAGS ::= BEGIN
 
-Request ::= CHOICE {
+ActivityType ::= ENUMERATED {
-	foo [0] NULL
+	reading(0),
+	exercises(1)
 }
 
-Response ::= CHOICE {
+Session ::= SEQUENCE {
-	msg [0] UTF8String
+	subject UTF8String,
+	type ActivityType,
+	timestamp INTEGER,
+	minutes INTEGER
+}
+
+RequestBody ::= CHOICE {
+	ping [0] NULL,
+	listActivities [1] NULL,
+	logSession [2] Session
+}
+
+Request ::= SEQUENCE {
+	transactionId INTEGER,
+	body RequestBody
+}
+
+Activity ::= SEQUENCE {
+	subject UTF8String,
+	type ActivityType,
+	priority INTEGER
+}
+
+Error ::= ENUMERATED {
+	invalidRequest(0),
+	invalidArguments(1),
+	serverError(2)
+}
+
+ResponseBody ::= CHOICE {
+	error [0] Error,
+	ack [1] NULL,
+	activities [2] SEQUENCE OF Activity
+}
+
+Response ::= SEQUENCE {
+	transactionId INTEGER,
+	body ResponseBody
 }
 
 END

server/src/session_server.erl

@@ -13,7 +13,8 @@ start_link(Socket) ->
 
 init(Socket) ->
 	ok = ssl:setopts(Socket, [{active, true}]),
-	{ok, #{socket => Socket}}.
+	process_flag(trap_exit, true),
+	{ok, #{socket => Socket, transactions => #{}}}.
 
 handle_call(_Request, _From, State) ->
 	{reply, ok, State}.
@@ -21,17 +22,28 @@ handle_call(_Request, _From, State) ->
 handle_cast(_Msg, State) ->
 	{noreply, State}.
 
-handle_info({ssl, Socket, Data}, State) ->
+handle_info({ssl, Socket, Data}, State = #{transactions := Transactions}) ->
 	case 'StudySystemProtocol':decode('Request', Data) of
-		{ok, {foo, _}} ->
+		{ok, {'Request', TransactionId, RequestBody}} ->
-			{ok, Encoded}
+			Pid = spawn_link(fun() -> handle_request(RequestBody) end),
-				= 'StudySystemProtocol':encode('Response', {msg, "Foo"}),
+			NewTransactions = maps:put(Pid, TransactionId, Transactions),
-			ok = ssl:send(Socket, Encoded);
+			{noreply, State#{transactions := NewTransactions}};
-		Result ->
+		{error, {asn1, _Reason}} ->
-			io:format("Invalid message: ~p~n", [Result]),
+			send(Socket, -1, {error, invalidRequest}),
-			ok
+			{noreply, State}
+	end;
+handle_info({'EXIT', Pid, Reason},
+			State = #{socket := Socket, transactions := Transactions})
+  when is_map_key(Pid, Transactions) ->
+	TransactionId = maps:get(Pid, Transactions),
+	Response = case Reason of
+				   {response, Value} -> Value;
+				   _ ->
+					   io:format("Error handling request: ~p~n", [Reason]),
+					   {error, serverError}
 			   end,
-	{noreply, State};
+	send(Socket, TransactionId, Response),
+	{noreply, State#{transactions := maps:remove(Pid, Transactions)}};
 handle_info({ssl_closed, _Socket}, State) ->
     {stop, normal, State};
 handle_info({ssl_error, _Socket, _Reason}, State) ->
@@ -44,3 +56,25 @@ terminate(_Reason, #{socket := Socket}) ->
 
 code_change(_OldVsn, State, _Extra) ->
 	{ok, State}.
+
+handle_request(Request) ->
+	timer:kill_after(500),
+	exit(map_request(Request)).
+
+map_request({ping, 'NULL'}) ->
+	{response, {ack, 'NULL'}};
+map_request({logSession, {'Session', Subject, Type, Timestamp, Minutes}}) ->
+	Session = {unicode:characters_to_list(Subject),
+			   Type, Timestamp, Minutes},
+	case subject_router:log_session(Session) of
+		ok -> {response, {ack, 'NULL'}};
+		{error, invalid_subject} -> {response, {error, invalidArguments}}
+	end;
+map_request(_) ->
+	{response, {error, invalidArguments}}.
+
+send(Socket, TransactionId, Response) ->
+	{ok, Encoded} = 'StudySystemProtocol':encode(
+					  'Response',
+					  {'Response', TransactionId, Response}),
+	ok = ssl:send(Socket, Encoded).

server/src/session_sup.erl

@@ -11,8 +11,8 @@ start_link() ->
 
 init([]) ->
 	SupFlags = #{strategy => simple_one_for_one,
-				 intensity => 10,
+				 intensity => 5,
-				 period => 1},
+				 period => 10},
 	ChildSpec = #{id => session_server,
 				  start => {session_server, start_link, []},
 				  restart => temporary,

server/src/study_system_server_app.erl

@@ -9,7 +9,11 @@
 start(_StartType, _StartArgs) ->
 	Port = application:get_env(study_system_server, port, 12888),
 	{ok, CertDir} = application:get_env(study_system_server, cert_dir),
-	proto_sup:start_link(Port, CertDir).
+	{ok, _Pid} = pg:start(study_system_server),
+	study_system_server_sup:start_link(Port, CertDir),
+	subject_sup:start_subject("Cybernetics"),
+	subject_sup:start_subject("Physics"),
+	subject_sup:start_subject("Linguistics").
 
 stop(_State) ->
 	ok.

server/src/study_system_server_sup.erl

@@ -0,0 +1,35 @@
+% Copyright (c) Camden Dixie O'Brien
+% SPDX-License-Identifier: AGPL-3.0-only
+
+-module(study_system_server_sup).
+-behaviour(supervisor).
+
+-export([start_link/2]).
+-export([init/1]).
+
+start_link(Port, CertDir) ->
+	supervisor:start_link({local, ?MODULE}, ?MODULE, [Port, CertDir]).
+
+init([Port, CertDir]) ->
+	SupFlags = #{strategy => one_for_one,
+				 intensity => 5,
+				 period => 10},
+	ChildSpecs = [#{id => subject_router,
+					start => {subject_router, start_link, []},
+					restart => permanent,
+					shutdown => 5000,
+					type => worker,
+					modules => [subject_router]},
+				  #{id => subject_sup,
+					start => {subject_sup, start_link, []},
+					restart => permanent,
+					shutdown => 5000,
+					type => supervisor,
+					modules => [subject_sup]},
+				  #{id => proto_sup,
+					start => {proto_sup, start_link, [Port, CertDir]},
+					restart => permanent,
+					shutdown => 5000,
+					type => supervisor,
+					modules => [proto_sup]}],
+	{ok, {SupFlags, ChildSpecs}}.

server/src/subject_router.erl

@@ -0,0 +1,75 @@
+% Copyright (c) Camden Dixie O'Brien
+% SPDX-License-Identifier: AGPL-3.0-only
+
+-module(subject_router).
+-behaviour(gen_server).
+
+-export([start_link/0, log_session/1]).
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
+		 terminate/2, code_change/3]).
+
+start_link() ->
+	gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
+
+log_session(Session) ->
+	gen_server:call(?MODULE, {log_session, Session}).
+
+init([]) ->
+	{MonitorRef, Pids} = pg:monitor(study_system_server, subject_servers),
+	SubjectTable = ets:new(subject_table, [private]),
+	register_servers(SubjectTable, Pids),
+	{ok, #{monitor_ref => MonitorRef,
+		   subject_table => SubjectTable}}.
+
+handle_call({log_session, {Subject, Type, Timestamp, Minutes}},
+			_From, State = #{subject_table := SubjectTable}) ->
+	case ets:lookup(SubjectTable, Subject) of
+		[{Subject, Pid}] ->
+			Pid ! {new_session, Type, Timestamp, Minutes},
+			{reply, ok, State};
+		[] ->
+			{reply, {error, invalid_subject}, State}
+	end;
+handle_call(_Request, _From, State) ->
+	{reply, ok, State}.
+
+handle_cast(_Msg, State) ->
+	{noreply, State}.
+
+handle_info({Ref, join, subject_servers, Pids},
+			State = #{monitor_ref := MonitorRef,
+					  subject_table := SubjectTable})
+  when Ref =:= MonitorRef ->
+	register_servers(SubjectTable, Pids),
+	{noreply, State};
+handle_info({Ref, leave, subject_servers, Pids},
+			State = #{monitor_ref := MonitorRef,
+					  subject_table := SubjectTable})
+  when Ref =:= MonitorRef ->
+	deregister_servers(SubjectTable, Pids),
+	{noreply, State};
+handle_info(_Info, State) ->
+	{noreply, State}.
+
+terminate(_Reason, _State) ->
+	ok.
+
+code_change(_OldVsn, State, _Extra) ->
+	{ok, State}.
+
+register_servers(SubjectTable, Pids) ->
+	[register_server(SubjectTable, Pid) || Pid <- Pids].
+
+register_server(SubjectTable, Pid) ->
+	try
+		{subject, Subject} = gen_server:call(Pid, get_subject),
+		ets:insert(SubjectTable, {Subject, Pid})
+	catch
+		_:_ -> ok
+	end.
+
+deregister_servers(SubjectTable, Pids) ->
+	[deregister_server(SubjectTable, Pid) || Pid <- Pids].
+
+deregister_server(SubjectTable, Pid) ->
+	ets:match_delete(SubjectTable, {'_', Pid}).

server/src/subject_server.erl

@@ -0,0 +1,39 @@
+% Copyright (c) Camden Dixie O'Brien
+% SPDX-License-Identifier: AGPL-3.0-only
+
+-module(subject_server).
+-behaviour(gen_server).
+
+-export([start_link/1]).
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
+		 terminate/2, code_change/3]).
+
+start_link(Subject) ->
+	gen_server:start_link(?MODULE, Subject, []).
+
+init(Subject) ->
+	pg:join(study_system_server, subject_servers, self()),
+	{ok, #{subject => Subject}}.
+
+handle_call(get_subject, _From, State = #{subject := Subject}) ->
+	{reply, {subject, Subject}, State};
+handle_call(_Request, _From, State) ->
+	{reply, ok, State}.
+
+handle_cast(_Msg, State) ->
+	{noreply, State}.
+
+handle_info({new_session, Type, Timestamp, Minutes},
+			State = #{subject := Subject}) ->
+	io:format(
+	  "Received new ~p session: type ~p, timestamp ~p, minutes ~p~n",
+	  [Subject, Type, Timestamp, Minutes]),
+	{noreply, State};
+handle_info(_Info, State) ->
+	{noreply, State}.
+
+terminate(_Reason, _State) ->
+	ok.
+
+code_change(_OldVsn, State, _Extra) ->
+	{ok, State}.

server/src/subject_sup.erl

@@ -0,0 +1,25 @@
+% Copyright (c) Camden Dixie O'Brien
+% SPDX-License-Identifier: AGPL-3.0-only
+
+-module(subject_sup).
+-behaviour(supervisor).
+
+-export([start_link/0, init/1, start_subject/1]).
+
+start_link() ->
+	supervisor:start_link({local, ?MODULE}, ?MODULE, []).
+
+init([]) ->
+	SupFlags = #{strategy => simple_one_for_one,
+				 intensity => 5,
+				 period => 10},
+	ChildSpec = #{id => subject_server,
+				  start => {subject_server, start_link, []},
+				  restart => permanent,
+				  shutdown => 5000,
+				  type => worker,
+				  modules => [subject_server]},
+	{ok, {SupFlags, [ChildSpec]}}.
+
+start_subject(Subject) ->
+	supervisor:start_child(?MODULE, [Subject]).