Implement transaction management in client

client/src/activities_view.vala

@@ -21,7 +21,7 @@ namespace StudySystemClient {
 	}
 
 	public class ActivitiesView : Gtk.Box {
-		public ActivitiesView()  {
+		public ActivitiesView(Connection connection)  {
 			margin_top = margin_bottom = margin_start = margin_end = 0;
 
 			var scrolled_window = new Gtk.ScrolledWindow();
@@ -44,8 +44,10 @@ 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(connection, activity);
+				card_container.append(card);
+			}
 
 			scrolled_window.set_child(card_container);
 			this.append(scrolled_window);
@@ -53,7 +55,7 @@ namespace StudySystemClient {
 	}
 
 	private class ActivityCard : Gtk.Frame {
-		public ActivityCard(Activity activity) {
+		public ActivityCard(Connection connection, Activity activity) {
 			add_css_class("card");
 
 			var content = new Gtk.Box(Gtk.Orientation.HORIZONTAL, 12);
@@ -82,7 +84,7 @@ namespace StudySystemClient {
 
 			set_child(content);
 
-			var log_session_popover = new LogSessionPopover();
+			var log_session_popover = new LogSessionPopover(connection);
 			log_session_popover.set_parent(button);
 			button.clicked.connect(() => log_session_popover.popup());
 		}
@@ -92,8 +94,11 @@ namespace StudySystemClient {
 		private const int DEFAULT_LENGTH = 30;
 
 		private Gtk.SpinButton input;
+		private Connection connection;
+
+		public LogSessionPopover(Connection connection) {
+			this.connection = connection;
 
-		public LogSessionPopover() {
 			var content = new Gtk.Box(Gtk.Orientation.HORIZONTAL, 6);
 
 			var label = new Gtk.Label("Minutes");
@@ -118,9 +123,11 @@ namespace StudySystemClient {
 			closed.connect(reset);
 		}
 
-		private void submit() {
+		private async void submit() {
 			reset();
 			popdown();
+			yield connection.send(new Request.Ping());
+			stderr.printf("Got ACK\n");
 		}
 
 		private void reset() {

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);
 		}
 	}
 

client/src/main_window.vala

@@ -16,7 +16,7 @@ namespace StudySystemClient {
 			header_bar.title_widget = title;
 			set_titlebar(header_bar);
 
-			var activities_view = new ActivitiesView();
+			var activities_view = new ActivitiesView(connection);
 			set_child(activities_view);
 		}
 	}

client/src/meson.build

@@ -14,6 +14,8 @@ lib = library(
     'connection.vala',
     'der.vala',
     'main_window.vala',
+    'request.vala',
+    'response.vala',
     'session_manager.vala',
   ) + resources,
   dependencies: [gtk_dep],

client/src/request.vala

@@ -0,0 +1,30 @@
+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_PRIORITIZED_ACTIVITIES = 1,
+			LOG_SESSION = 2,
+		}
+
+		internal Der.Datum datum;
+	}
+
+	public class Ping : Body {
+		public Ping() {
+			datum = new Der.Choice(Tag.PING, new Der.Null());
+		}
+	}
+}

client/src/response.vala

@@ -0,0 +1,98 @@
+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(
+					"Response sequnce contained %u fields (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 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");
+			}
+		}
+	}
+}

client/src/session_manager.vala

@@ -1,6 +1,6 @@
 namespace StudySystemClient {
 	public class SessionManager {
-		public delegate void ReceiveCallback(uint8[] msg);
+		public delegate void ReceiveCallback(owned uint8[] msg);
 
 		private const uint INIT_RECONNECT_WAIT_MS = 500;
 		private const uint MAX_RECONNECT_WAIT_MS = 60000;
@@ -45,8 +45,7 @@ namespace StudySystemClient {
 		private void try_start_session() {
 			try {
 				session = session_factory.start_session();
-				session.received.connect(
+				session.received.connect((msg) => receive_callback(msg));
-					(msg) => receive_callback(msg));
 				reconnect_wait_ms = INIT_RECONNECT_WAIT_MS;
 			} catch (Error _) {
 				Thread.usleep(1000 * reconnect_wait_ms);