git » sdk » commit b7e106e

diff --git a/borogove/ChannelPinger.hx b/borogove/ChannelPinger.hx
new file mode 100644
index 0000000..872f350
--- /dev/null
+++ b/borogove/ChannelPinger.hx
@@ -0,0 +1,70 @@
+package borogove;
+
+import borogove.Chat;
+
+using Lambda;
+
+private typedef ScheduledChannel = {
+	final channel: Channel;
+	var deadline: Float;
+}
+
+class ChannelPinger {
+	private static inline final PING_INTERVAL_MS = 5 * 60 * 1000;
+	private static inline final CHECK_INTERVAL_MS = 60 * 1000;
+	private static inline final PING_TRIGGER_WINDOW_MS = 30 * 1000;
+	private static inline final COALESCE_WINDOW_MS = PING_TRIGGER_WINDOW_MS * 2;
+
+	private final now: ()->Float;
+	private final scheduledChannels = new Map<String, ScheduledChannel>();
+	private var timer: Null<haxe.Timer> = null;
+
+	public function new(?now: ()->Float) {
+		this.now = now ?? (() -> haxe.Timer.stamp() * 1000);
+	}
+
+	public function startTimer(): Void {
+		if (timer != null) throw "ChannelPinger timer already exists";
+		timer = new haxe.Timer(CHECK_INTERVAL_MS);
+		timer.run = pingPastDue;
+	}
+
+	public function stopTimer(): Void {
+		timer?.stop();
+		timer = null;
+	}
+
+	public function schedule(channel: Channel, ?deadline: Float): Void {
+		if (channel.uiState != Open) return;
+
+		scheduledChannels.set(channel.chatId, {
+			channel: channel,
+			deadline: deadline ?? now() + PING_INTERVAL_MS,
+		});
+	}
+
+	public function remove(channel: Channel): Void {
+		scheduledChannels.remove(channel.chatId);
+	}
+
+	public function pingPastDue(): Void {
+		pingDueBy(now());
+	}
+
+	public function pingDueInWindow(): Void {
+		final currentTime = now();
+		if (getDueBy(currentTime + PING_TRIGGER_WINDOW_MS).length > 0) {
+			pingDueBy(currentTime + COALESCE_WINDOW_MS);
+		}
+	}
+
+	private function pingDueBy(cutoff: Float): Void {
+		final due = getDueBy(cutoff);
+
+		for (scheduled in due) scheduled.channel.selfPing(false);
+	}
+
+	private function getDueBy(cutoff: Float): Array<ScheduledChannel> {
+		return scheduledChannels.filter(sc -> sc.deadline <= cutoff);
+	}
+}
diff --git a/borogove/Chat.hx b/borogove/Chat.hx
index 4d07487..149ec92 100644
--- a/borogove/Chat.hx
+++ b/borogove/Chat.hx
@@ -1511,6 +1511,7 @@ class Channel extends Chat {
 		if (uiState == Invited) return;
 
 		if (uiState == Closed) {
+			client.channelPinger.remove(this);
 			client.sendPresence(
 				getFullJid().asString(),
 				(stanza) -> {
@@ -1585,10 +1586,7 @@ class Channel extends Chat {
 			final desiredFullJid = JID.parse(chatId).withResource(client.displayName());
 			client.sendPresence(desiredFullJid.asString());
 		}
-		// We did a self ping to see if we were in the room and found we are
-		// But we may have missed messages if we were disconnected in the middle
-		inSync = false;
-		persistence.syncPoint(client.accountId(), chatId).then(point -> doSync(point));
+		client.channelPinger.schedule(this);
 	}
 
 	override public function getDisplayName() {
@@ -1817,6 +1815,7 @@ class Channel extends Chat {
 			} else {
 				self = member;
 				outbox.start();
+				client.channelPinger.schedule(this);
 			}
 			if (!noStore) client.trigger("chats/update", [this]);
 		}
diff --git a/borogove/Client.hx b/borogove/Client.hx
index 4eb44d3..d58868b 100644
--- a/borogove/Client.hx
+++ b/borogove/Client.hx
@@ -129,6 +129,8 @@ class Client extends EventEmitter {
 	@:allow(borogove)
 	private var inSync(default, null) = false;
 	private var firstSync = true;
+	@:allow(borogove.Channel)
+	private final channelPinger: ChannelPinger;
 
 	/**
 		Create a new Client to connect to a particular account
@@ -155,8 +157,10 @@ class Client extends EventEmitter {
 		if (SignalProtocol.exists()) this.omemo = new OMEMO(this, persistence);
 #end
 		stream = new Stream();
+		channelPinger = new ChannelPinger();
 		stream.on("status/online", this.onConnected);
 		stream.on("status/offline", (data) -> {
+			channelPinger.stopTimer();
 			this.trigger("status/offline", {});
 		});
 
@@ -479,6 +483,10 @@ class Client extends EventEmitter {
 			final channel = Std.downcast(chat, Channel);
 			if (channel != null) channel.selfPing(true);
 		}
+		if (stanza.attr.get("type") != "error" && from != null) {
+			final channel = Std.downcast(getChat(from.asBare().asString()), Channel);
+			if (channel != null) channelPinger.schedule(channel);
+		}
 
 		var newChat: Null<Chat> = null;
 
@@ -900,6 +908,7 @@ class Client extends EventEmitter {
 		@param completely if true chats, messages, etc will be deleted as well
 	**/
 	public function logout(completely: Bool) {
+		channelPinger.stopTimer();
 		persistence.removeAccount(accountId(), completely);
 		final disable = new Push2Disable(jid.asBare().asString());
 		disable.onFinished(() -> {
@@ -1097,6 +1106,7 @@ class Client extends EventEmitter {
 			}
 
 			stream.emitSMupdates = true;
+			channelPinger.startTimer();
 			this.trigger("status/online", {});
 			this.trigger("chats/update", chats);
 			return EventHandled;
@@ -1150,6 +1160,7 @@ class Client extends EventEmitter {
 						sendPresence();
 						joinAllChannels();
 					}
+					channelPinger.startTimer();
 					this.trigger("status/online", {});
 					trace("SYNC: done");
 				});
@@ -2063,6 +2074,7 @@ class Client extends EventEmitter {
 
 	@:allow(borogove)
 	private function sendStanza(stanza:Stanza) {
+		channelPinger.pingDueInWindow();
 		if (stanza.attr.get("id") == null) stanza.attr.set("id", ID.unique());
 		stream.sendStanza(stanza);
 	}
diff --git a/borogove/Map.js.hx b/borogove/Map.js.hx
index f51fd05..de209aa 100644
--- a/borogove/Map.js.hx
+++ b/borogove/Map.js.hx
@@ -40,6 +40,10 @@ abstract Map<K,V>(NativeMap<K,V>) {
 		return [for (x in this) f(x)];
 	}
 
+	public inline function filter(f: V->Bool):Array<V> {
+		return [for (x in this) if (f(x)) x];
+	}
+
 	public inline function keyValueIterator():KeyValueIterator<K, V> {
 		return new HaxeKVIterator(this.entries());
 	}
diff --git a/borogove/OMEMO.hx b/borogove/OMEMO.hx
index 0664668..6d76637 100644
--- a/borogove/OMEMO.hx
+++ b/borogove/OMEMO.hx
@@ -834,7 +834,7 @@ class OMEMO {
 		return promPayload;
 	}
 
-	private function sendKeyExchange(deviceId:Int, addr:SignalProtocolAddress) {
+	private function sendKeyExchange(deviceId:Int, addr:SignalProtocolAddress):Promise<Stanza> {
 		trace("OMEMO: Preparing key exchange stanza...");
 		final emptyPayload = Bytes.alloc(32).toString();
 		final promEncryptedMessage = encryptPayloadWithNewKey(emptyPayload);
diff --git a/test/TestAll.hx b/test/TestAll.hx
index 2cffc25..73384e9 100644
--- a/test/TestAll.hx
+++ b/test/TestAll.hx
@@ -25,6 +25,7 @@ class TestAll {
 			new TestJID(),
 			new TestMember(),
 			new TestMemberUpdate(),
+			new TestChannelPinger(),
 			new TestMucSettingsCommand(),
 			new TestPresence(),
 			new TestReaction(),
diff --git a/test/TestChannelPinger.hx b/test/TestChannelPinger.hx
new file mode 100644
index 0000000..8478ed8
--- /dev/null
+++ b/test/TestChannelPinger.hx
@@ -0,0 +1,158 @@
+package test;
+
+import borogove.ChannelPinger;
+import borogove.Chat.Channel;
+import borogove.Chat.UiState;
+import borogove.EventEmitter.EventResult;
+import borogove.JID;
+import borogove.Stanza;
+import borogove.persistence.Dummy;
+import utest.Assert;
+
+@:access(borogove)
+class TestChannelPinger extends utest.Test {
+	private var client: borogove.Client;
+
+	public function setup() {
+		client = new borogove.Client("test@example.com", new Dummy());
+	}
+
+	public function testStartTimerErrorsIfCalledTwice() {
+		final pinger = new ChannelPinger();
+		pinger.startTimer();
+		Assert.raises(pinger.startTimer);
+	}
+
+	public function testScheduleAddsChannel() {
+		final now = 1000.0;
+		final channel = channel(client, "room@example.com");
+		final pinger = new ChannelPinger(() -> now);
+
+		Assert.isNull(pinger.scheduledChannels[channel.chatId]);
+
+		pinger.schedule(channel);
+
+		final scheduled = pinger.scheduledChannels[channel.chatId];
+		Assert.equals(channel, scheduled.channel);
+		Assert.equals(now + ChannelPinger.PING_INTERVAL_MS, scheduled.deadline);
+	}
+
+	public function testScheduleUpdatesDeadline() {
+		var now = 1000.0;
+		final channel = channel(client, "room@example.com");
+		final pinger = new ChannelPinger(() -> now);
+
+		pinger.schedule(channel);
+		Assert.equals(
+			now + ChannelPinger.PING_INTERVAL_MS,
+			pinger.scheduledChannels[channel.chatId].deadline,
+		);
+
+		now += 1000;
+
+		pinger.schedule(channel);
+		Assert.equals(
+			now + ChannelPinger.PING_INTERVAL_MS,
+			pinger.scheduledChannels[channel.chatId].deadline,
+		);
+	}
+
+	public function testScheduleAcceptsCustomDeadline() {
+		final channel = channel(client, "room@example.com");
+		final pinger = new ChannelPinger();
+		final deadline = 5000;
+
+		pinger.schedule(channel, deadline);
+
+		Assert.equals(
+			deadline,
+			pinger.scheduledChannels[channel.chatId].deadline,
+		);
+	}
+
+	public function testScheduleOnlySchedulesOpenChannels() {
+		final channel = channel(client, "room@example.com", Closed);
+		final pinger = new ChannelPinger();
+		pinger.schedule(channel);
+
+		Assert.isNull(pinger.scheduledChannels[channel.chatId]);
+	}
+
+	public function testRemoveRemovesScheduledChannel() {
+		final channel = channel(client, "room@example.com");
+		final pinger = new ChannelPinger();
+
+		pinger.schedule(channel);
+		Assert.notNull(pinger.scheduledChannels[channel.chatId]);
+
+		pinger.remove(channel);
+		Assert.isNull(pinger.scheduledChannels[channel.chatId]);
+	}
+
+	public function testPingPastDue() {
+		final now = 1000.0;
+		final pastDue = channel(client, "room1@example.com");
+		final dueNow = channel(client, "room2@example.com");
+		final dueInFuture = channel(client, "room3@example.com");
+		final pinger = new ChannelPinger(() -> now);
+
+		pinger.schedule(pastDue, now - 1);
+		pinger.schedule(dueNow, now);
+		pinger.schedule(dueInFuture, now + 1);
+
+		final pinged = capturePings(pinger.pingPastDue);
+
+		assertSameList([pastDue.chatId, dueNow.chatId], pinged);
+	}
+
+	public function testPingDueInWindowSendsEverythingIn60SecondsIfSomeIn30Seconds() {
+		final now = 1000.0;
+		final dueInNext30 = channel(client, "room1@example.com");
+		final dueInNext60 = channel(client, "room2@example.com");
+		final dueGreaterThan60 = channel(client, "room3@example.com");
+		final pinger = new ChannelPinger(() -> now);
+
+		pinger.schedule(dueInNext30, now + 30 * 1000);
+		pinger.schedule(dueInNext60, now + 60 * 1000);
+		pinger.schedule(dueGreaterThan60, now + 60 * 1000 + 1);
+
+		final pinged = capturePings(pinger.pingDueInWindow);
+
+		assertSameList([dueInNext30.chatId, dueInNext60.chatId], pinged);
+	}
+
+	public function testPingDueInWindowDoesNothingIfNothingDueInNext30() {
+		final now = 1000.0;
+		final room = channel(client, "room@example.com");
+		final pinger = new ChannelPinger(() -> now);
+		pinger.schedule(room, now + 30 * 1000 + 1);
+
+		final pinged = capturePings(pinger.pingDueInWindow);
+
+		Assert.same([], pinged);
+	}
+
+	private function channel(client, id: String, state: UiState = Open): Channel {
+		return new Channel(client, client.stream, client.persistence, id, state);
+	}
+
+	private function capturePings(operation: Void->Void): Array<String> {
+		final pinged = [];
+		client.stream.on("sendStanza", (stanza: Stanza) -> {
+			if (stanza.name == "iq" && stanza.getChild("ping", "urn:xmpp:ping") != null) {
+				pinged.push(JID.parse(stanza.attr.get("to")).asBare().asString());
+			}
+			return EventHandled;
+		});
+
+		operation();
+
+		return pinged;
+	}
+
+	private function assertSameList<T>(expected: Array<T>, actual: Array<T>) {
+		expected.sort(Reflect.compare);
+		actual.sort(Reflect.compare);
+		Assert.same(expected, actual);
+	}
+}