### ----------------------------------------------------
### -- Protocol::MQTT
### -- Licenses: Artistic-2.0
### -- Authors: Leon Timmermans
### -- File: t/04-client.rakutest
### ----------------------------------------------------
#! raku

use v6.d;

use Test;

use Protocol::MQTT :qos;

my $client-identifier = 'Test';
my $keep-alive-interval = 29;
my $resend-interval = 7;
my $client = Protocol::MQTT::Client.new(:$client-identifier, :$keep-alive-interval, :$resend-interval);

my $now = now;

subtest 'Connecting', {
	is($client.state, Protocol::MQTT::Disconnected, 'state=Disconnected');
	is($client.next-expiration(), Instant);
	my $promise = $client.connect;
	is($client.state, Protocol::MQTT::Unconnected, 'state=Unconnected');
	is($promise.status, Planned, 'Promise is planned');
	is($client.next-expiration(), Instant);

	my @messages = $client.next-events($now);
	is($client.state, Protocol::MQTT::Connecting, 'state=Connecting');
	is(+@messages, 1, 'Sending one message');
	is($promise.status, Planned, 'Promise is stil planned');
	is($client.next-expiration(), $now + $resend-interval);
	my $expected = Protocol::MQTT::Packet::Connect.new(:$client-identifier, :$keep-alive-interval);
	is-deeply(@messages[0], $expected, 'Packet is as expected');

	$client.received-packet(Protocol::MQTT::Packet::ConnAck.new, ++$now);
	is($promise.status, Kept, 'Promise is kept');
	is($client.state, Protocol::MQTT::Connected, 'state=Connected');
}

subtest 'Sending Qos=At-most-once', {
	my $topic = 'foo/bar';
	my $payload = 'whatever'.encode;
	my $qos = At-most-once;

	my $promise = $client.publish($topic, $payload, $qos, False, $now);
	is($promise.status, Kept, 'Promise is kept');

	my @messages = $client.next-events($now);
	is(+@messages, 1, 'Sending one message');

	my $expected = Protocol::MQTT::Packet::Publish.new(:$topic, :$payload, :$qos);
	is-deeply(@messages[0], $expected, 'Packet is as expected');

	is($client.next-expiration, $now + $keep-alive-interval, 'Next expiration at keep-alive');
}

subtest 'Sending Qos=At-least-once', {
	my $packet-id;
	my $topic = 'foo/bar';
	my $payload = 'whatever'.encode;
	my $qos = At-least-once;

	my $promise = $client.publish($topic, $payload, $qos, False, ++$now);

	subtest 'Original publish', {
		is($promise.status, Planned, 'Promise is not immediate');

		my @messages = $client.next-events($now);
		is(+@messages, 1, 'Sending one message (1)');

		given @messages[0] -> $message {
			isa-ok($message, Protocol::MQTT::Packet::Publish);
			$packet-id = $message.packet-id;
			ok($packet-id, 'Packet id is present');
			my $expected = Protocol::MQTT::Packet::Publish.new(:$topic, :$payload, :$qos, :$packet-id);
			is-deeply($message, $expected, 'Packet is as expected');
		}

		is($client.next-expiration, $now + $resend-interval, 'Next expiration is at resend');
		is-deeply($client.next-events($now), [], 'No events expected');
		is($client.next-expiration, $now + $resend-interval, 'Next expiration is at resend');

		is($client.pending-acknowledgements, 1, 'Has 1 pending message');
	}

	subtest 'Resend publish', {
		$now += $resend-interval;

		my @messages = $client.next-events($now);
		is($promise.status, Planned, 'Promise is still not true');
		is(+@messages, 1, 'Got one message (2)');
		given @messages[0] -> $message {
			isa-ok($message, Protocol::MQTT::Packet::Publish, 'Packet is a publish');
			is($message.packet-id, $packet-id, 'Packet id has expected value');
			my $expected = Protocol::MQTT::Packet::Publish.new(:$topic, :$payload, :$qos, :dup, :$packet-id);
			is-deeply($message, $expected, 'Packet is as expected');
		}

		is($client.pending-acknowledgements, 1, 'Has 1 pending message');
		is($client.next-expiration, $now + $resend-interval, 'Next expiration is at resend');
	}

	subtest 'Receive Acknowledgement', {
		$now++;
		is($promise.status, Planned, 'Promise is still not true');
		$client.received-packet(Protocol::MQTT::Packet::PubAck.new(:$packet-id), $now);
		is($promise.status, Kept, 'Promise is still finally true');
		my @messages3 = $client.next-events($now);
		is(+@messages3, 0, 'Got no messages');

		is($client.pending-acknowledgements, 0, 'Has 0 pending message');
		is($client.next-expiration, $now + $keep-alive-interval, 'Next expiration is at keep-alive');

		$now += $resend-interval;
		is-deeply($client.next-events($now), [], 'No events expected');
	}
}

subtest 'Sending Qos=Exactly-once', {
	my $packet-id;
	my $topic = 'foo/bar';
	my $payload = 'whatever'.encode;
	my $qos = Exactly-once;

	$now++;
	my $promise = $client.publish($topic, $payload, $qos, False, $now);

	subtest 'Original publish', {
		is($promise.status, Planned, 'Promise is not immediate');

		my @messages = $client.next-events($now);
		is(+@messages, 1, 'Sending one message');

		given @messages[0] -> $message {
			isa-ok($message, Protocol::MQTT::Packet::Publish);
			$packet-id = $message.packet-id;
			ok($packet-id, 'Packet id is present');
			my $expected = Protocol::MQTT::Packet::Publish.new(:$topic, :$payload, :$qos, :$packet-id);
			is-deeply($message, $expected, 'Packet is as expected');
		}

		is-deeply($client.next-events($now), [], 'No events expected');
		is($client.next-expiration, $now + $resend-interval, 'Next expiration is at resend');

		is($client.pending-acknowledgements, 1, 'Has 1 pending message');
	}

	subtest 'Resend publish', {
		$now += $resend-interval;

		my @messages = $client.next-events($now);
		is(+@messages, 1, 'Sending one message');
		given @messages[0] -> $message {
			isa-ok($message, Protocol::MQTT::Packet::Publish, 'Packet is a publish');
			is($message.packet-id, $packet-id, 'Packet id has expected value');
			my $expected = Protocol::MQTT::Packet::Publish.new(:$topic, :$payload, :$qos, :!retain, :dup, :$packet-id);
			is-deeply($message, $expected, 'Packet is as expected');
		}

		is($promise.status, Planned, 'Promise is still not true');
		is($client.pending-acknowledgements, 1, 'Has 1 pending message');
		is($client.next-expiration, $now + $resend-interval, 'Next expiration is at resend');
	}

	subtest 'Receive record, send release', {
		$now++;
		$client.received-packet(Protocol::MQTT::Packet::PubRec.new(:$packet-id), $now);

		my @messages = $client.next-events($now);
		is(+@messages, 1, 'Sending 1 message');
		is-deeply(@messages[0], Protocol::MQTT::Packet::PubRel.new(:$packet-id), 'PubRel is as expected');

		is($client.pending-acknowledgements, 1, 'Has 1 pending message');
		is-deeply($client.next-events($now), [], 'No events expected');
		is($client.next-expiration, $now + $resend-interval, 'Next expiration is at resend');
	}

	subtest 'Resend release', {
		$now += $resend-interval;

		my @messages = $client.next-events($now);
		is(+@messages, 1, 'Sending 1 message');
		is-deeply(@messages[0], Protocol::MQTT::Packet::PubRel.new(:$packet-id), 'PubRel is as expected');

		is($client.pending-acknowledgements, 1, 'Has 1 pending message');
		is-deeply($client.next-events($now), [], 'No events expected');

		$client.received-packet(Protocol::MQTT::Packet::PubComp.new(:$packet-id), $now);

		is($client.pending-acknowledgements, 0, 'Has no pending message');
		is-deeply($client.next-events($now), [], 'No events expected');
		is($client.next-expiration, $now + $keep-alive-interval, 'Next expiration is at resend');

		$now += $resend-interval;

		is-deeply($client.next-events($now), [], 'No events expected');
	}
}

subtest 'Keep-alive', {
	$client.received-packet(Protocol::MQTT::Packet::PingResp.new(), $now);
	is-deeply($client.next-events($now), [], 'No events expected');
	is($client.next-expiration, $now + $keep-alive-interval);

	$now += $keep-alive-interval;

	is-deeply($client.next-events($now), [Protocol::MQTT::Packet::PingReq.new], 'Ping request expected');
	is($client.next-expiration, $now + $keep-alive-interval);
	$client.received-packet(Protocol::MQTT::Packet::PingResp.new(), ++$now);
	is-deeply($client.next-events($now), [], 'No events expected');
	is($client.next-expiration, $now + $keep-alive-interval);
}

subtest 'Receiving Qos=At-most-once', {
	$now++;
	
	my @all;
	
	my $topic = '#';
	my $qos = At-most-once;
	my $packet-id;

	my $tap = $client.incoming.tap: -> $message { @all.push($message); }
	LEAVE { $tap.close };

	subtest 'Sending subscription', {
		$client.subscribe($topic, $qos, $now);
		my @messages = $client.next-events($now);
		is(+@messages, 1, 'Expecting one packet');
		given @messages[0] -> $message {
			$packet-id = $message.packet-id;
			is-deeply($message, Protocol::MQTT::Packet::Subscribe.new(:$topic, :$qos, :$packet-id), 'Sent subscription is as expected');
		}

		is-deeply($client.next-events($now), [], 'No events expected');
	}

	subtest 'Resending subscription', {
		$now += $resend-interval;
		my @messages = $client.next-events($now);
		is(+@messages, 1, 'Expecting one packet');
		is-deeply(@messages[0], Protocol::MQTT::Packet::Subscribe.new(:$topic, :$qos, :$packet-id), 'Resent subscription is as expected');
	}

	subtest 'Subscription acknowledged', {
		$now++;
		$client.received-packet(Protocol::MQTT::Packet::SubAck.new(:$packet-id, :qos-levels[ $qos ]), $now);
		
		$now += $resend-interval;
		is-deeply($client.next-events($now), [], 'No events expected');
	}

	subtest 'First message', {
		$now++;
		$client.received-packet(Protocol::MQTT::Packet::Publish.new(:2packet-id, :topic<foo>, :payload('Hello world'.encode), :$qos), $now);
		is-deeply($client.next-events($now), [], 'No events expected');

		is(+@all, 1, 'Received first publish');
		is-deeply(@all[0], Protocol::MQTT::Message.new(:topic<foo>, :payload('Hello world'.encode)), 'publish is "Hello world"');
	}

	subtest 'Second message', {
		$now++;

		$client.received-packet(Protocol::MQTT::Packet::Publish.new(:3packet-id, :topic<foo>, :payload('Hello again'.encode), :$qos), $now);
		is-deeply($client.next-events($now), [], 'No events expected');

		is(+@all, 2, 'Received second publish');
		is-deeply(@all[1], Protocol::MQTT::Message.new(:topic<foo>, :payload('Hello again'.encode)), 'publish is "Hello again"');
	}
}

subtest 'Receiving Qos=At-least-once', {
	$now++;
	
	my @all;
	
	my $topic = '#';
	my $qos = At-least-once;
	my $packet-id;

	my $tap = $client.incoming.tap: -> $message { @all.push($message); }
	LEAVE { $tap.close };

	subtest 'Sending subscription', {
		$client.subscribe($topic, $qos, $now);
		my @messages = $client.next-events($now);
		is(+@messages, 1, 'Expecting one packet');
		given @messages[0] -> $message {
			$packet-id = $message.packet-id;
			is-deeply($message, Protocol::MQTT::Packet::Subscribe.new(:$topic, :$qos, :$packet-id), 'Sent subscription is as expected');
		}

		is-deeply($client.next-events($now), [], 'No events expected');
	}

	subtest 'Resending subscription', {
		$now += $resend-interval;
		my @messages = $client.next-events($now);
		is(+@messages, 1, 'Expecting one packet');
		is-deeply(@messages[0], Protocol::MQTT::Packet::Subscribe.new(:$topic, :$qos, :$packet-id), 'Resent subscription is as expected');
	}

	subtest 'Subscription acknowledged', {
		$now++;
		$client.received-packet(Protocol::MQTT::Packet::SubAck.new(:$packet-id, :qos-levels[ $qos ]), $now);

		$now += $resend-interval;
		is-deeply($client.next-events($now), [], 'No events expected');
	}

	subtest 'First message', {
		$now++;
		$client.received-packet(Protocol::MQTT::Packet::Publish.new(:5packet-id, :topic<foo>, :payload('Hello world'.encode), :$qos), $now);
		is-deeply($client.next-events($now), [ Protocol::MQTT::Packet::PubAck.new(:5packet-id) ], 'Packet triggers ack');

		is(+@all, 1, 'Received first publish');
		my $expected = Protocol::MQTT::Message.new(:topic<foo>, :payload('Hello world'.encode), :qos(At-least-once));
		is-deeply(@all[0], $expected, 'publish is "Hello world"');
	}

	subtest 'Second message', {
		$now++;

		$client.received-packet(Protocol::MQTT::Packet::Publish.new(:6packet-id, :topic<foo>, :payload('Hello again'.encode), :$qos), $now);
		is-deeply($client.next-events($now), [ Protocol::MQTT::Packet::PubAck.new(:6packet-id) ], 'Packet triggers ack');

		is(+@all, 2, 'Received second publish');
		is-deeply(@all[1], Protocol::MQTT::Message.new(:topic<foo>, :payload('Hello again'.encode), :qos(At-least-once)), 'publish is "Hello again"');
	}
}

subtest 'Receiving Qos=Exactly-once', {
	$now++;
	
	my @all;
	
	my $topic = '#';
	my $qos = Exactly-once;
	my $packet-id;

	my $tap = $client.incoming.tap: -> $message { @all.push($message); }
	LEAVE { $tap.close };

	subtest 'Sending subscription', {
		$client.subscribe($topic, $qos, $now);
		my @messages = $client.next-events($now);
		is(+@messages, 1, 'Expecting one packet');
		given @messages[0] -> $message {
			$packet-id = $message.packet-id;
			is-deeply($message, Protocol::MQTT::Packet::Subscribe.new(:$topic, :$qos, :$packet-id), 'Sent subscription is as expected');
		}

		is-deeply($client.next-events($now), [], 'No events expected');
	}

	subtest 'Resending subscription', {
		$now += $resend-interval;
		my @messages = $client.next-events($now);
		is(+@messages, 1, 'Expecting one packet');
		is-deeply(@messages[0], Protocol::MQTT::Packet::Subscribe.new(:$topic, :$qos, :$packet-id), 'Resent subscription is as expected');
	}

	subtest 'Subscription acknowledged', {
		$now++;
		$client.received-packet(Protocol::MQTT::Packet::SubAck.new(:$packet-id, :qos-levels[ $qos ]), $now);

		$now += $resend-interval;
		is-deeply($client.next-events($now), [], 'No events expected');
	}

	subtest 'First message', {
		$now++;
		$client.received-packet(Protocol::MQTT::Packet::Publish.new(:8packet-id, :topic<foo>, :payload('Hello world'.encode), :$qos), $now);
		is-deeply($client.next-events($now), [ Protocol::MQTT::Packet::PubRec.new(:8packet-id) ], 'Packet triggers ack');

		is(+@all, 1, 'Received first publish');
		is-deeply(@all[0], Protocol::MQTT::Message.new(:topic<foo>, :payload('Hello world'.encode), :qos(Exactly-once)), 'publish is "Hello world"');

		$client.received-packet(Protocol::MQTT::Packet::Publish.new(:8packet-id, :topic<foo>, :payload('Hello world'.encode), :$qos), $now);
		is-deeply($client.next-events($now), [ Protocol::MQTT::Packet::PubRec.new(:8packet-id) ], 'Packet triggers ack');

		is(+@all, 1, 'Ignored repeat of publish');

		$client.received-packet(Protocol::MQTT::Packet::PubRel.new(:8packet-id), $now);
		is-deeply($client.next-events($now), [ Protocol::MQTT::Packet::PubComp.new(:8packet-id) ], 'Release trigger Complete');

		$client.received-packet(Protocol::MQTT::Packet::Publish.new(:8packet-id, :topic<foo>, :payload('Hello world'.encode), :qos(At-most-once)), $now);

		is(+@all, 2, 'After PubComp, packet-id may be used again');
	}

	subtest 'Second message', {
		$now++;

		$client.received-packet(Protocol::MQTT::Packet::Publish.new(:9packet-id, :topic<foo>, :payload('Hello again'.encode), :$qos), $now);
		is-deeply($client.next-events($now), [ Protocol::MQTT::Packet::PubRec.new(:9packet-id) ], 'Packet triggers ack');

		is(+@all, 3, 'Received second publish');
		is-deeply(@all[2], Protocol::MQTT::Message.new(:topic<foo>, :payload('Hello again'.encode), :qos(Exactly-once)), 'publish is "Hello again"');

		$client.received-packet(Protocol::MQTT::Packet::PubRel.new(:9packet-id), $now);
		is-deeply($client.next-events($now), [ Protocol::MQTT::Packet::PubComp.new(:9packet-id) ], 'Release trigger Complete');
	}
}

subtest 'Disconnect', {
	$now++;
	my $client2 = Protocol::MQTT::Client.new(:$client-identifier, :$keep-alive-interval, :$resend-interval);
	$client2.connect;
	is-deeply($client2.next-events($now), [ Protocol::MQTT::Packet::Connect.new(:$client-identifier, :$keep-alive-interval)]);
	$now++;
	$client2.received-packet(Protocol::MQTT::Packet::ConnAck.new, $now);
	is($client2.state, Protocol::MQTT::Connected, 'We are connected');

	$client2.disconnect;
	is-deeply($client2.next-events($now), [ Protocol::MQTT::Packet::Disconnect.new ], 'Sending a disconnect');
}

subtest 'Timeout', {
	$now++;
	is($client.state, Protocol::MQTT::Connected, 'We are still connected');

	$now += 2 * $keep-alive-interval;
	is-deeply($client.next-events($now), [], 'No more events');
	is($client.state, Protocol::MQTT::Disconnected, 'We ware now disconnected');
}

subtest 'Connect timeout', {
	$now++;
	my $client3 = Protocol::MQTT::Client.new(:$client-identifier, :$keep-alive-interval, :$resend-interval);
	is($client3.state, Protocol::MQTT::Disconnected, 'State starts as disconnected');
	my $promise = $client3.connect;
	is($promise.status, Planned, 'Promise is planned');
	is($client3.state, Protocol::MQTT::Unconnected, 'State starts as unconnected');
	is-deeply($client3.next-events($now), [ Protocol::MQTT::Packet::Connect.new(:$client-identifier, :$keep-alive-interval) ], 'First packet is connect');
	is($client3.state, Protocol::MQTT::Connecting, 'state is connected after sending connect package');
	$now += $resend-interval;
	is-deeply($client3.next-events(++$now), [], 'Don\'t have packets after connect timeout');
	is($client3.state, Protocol::MQTT::Unconnected, 'Unconnected after connect timeout');
	is($promise.status, Planned, 'Promise is planned');
	$client3.disconnect;
	is-deeply($client3.next-events(++$now), [], 'Don\'t have packets after disconnect');
	is($client3.state, Protocol::MQTT::Disconnected, 'Unconnected after disconnect');
	is($promise.status, Broken, 'Promise is broken');
}

subtest 'Reconnect', {
	my $client4 = Protocol::MQTT::Client.new(:$client-identifier, :$keep-alive-interval, :$resend-interval);
	is($client4.state, Protocol::MQTT::Disconnected, 'We are disconnected');
	$client4.connect;
	is($client4.state, Protocol::MQTT::Unconnected, 'State starts as unconnected');
	is-deeply($client4.next-events($now), [ Protocol::MQTT::Packet::Connect.new(:$client-identifier, :$keep-alive-interval) ], 'First packet is connect');
	$client4.received-packet(Protocol::MQTT::Packet::ConnAck.new, $now);
	is($client4.state, Protocol::MQTT::Connected, 'We are connected');
	$now++;
	my $supply = $client4.subscribe('foo/bar', At-most-once, $now);
	my @messages = $client4.next-events($now);
	is(+@messages, 1, 'Sending one message');
	isa-ok(@messages[0], Protocol::MQTT::Packet::Subscribe, 'Message is a Subscribe');
	$client4.received-packet(Protocol::MQTT::Packet::SubAck.new(:packet-id(@messages[0].packet-id)), $now);
	my $packet-id = @messages[0].packet-id;

	my $promise = $client4.connect;
	is($promise.status, Planned, 'Promise is planned');
	$now++;

	is($client4.state, Protocol::MQTT::Unconnected, 'We are unconnected');
	is-deeply($client4.next-events($now), [ Protocol::MQTT::Packet::Connect.new(:$client-identifier, :$keep-alive-interval) ], 'Next packet is second connect');
	$client4.received-packet(Protocol::MQTT::Packet::ConnAck.new, $now);
	is($client4.state, Protocol::MQTT::Connected, 'We are connected');
	my @messages2 = $client4.next-events($now);
	is(+@messages2, 1, 'Sending one message again');
	is-deeply(@messages2[0].subscriptions.list, Array[Protocol::MQTT::Packet::Subscribe::Subscription].new(Packet::Subscribe::Subscription.new(:topic<foo/bar>, :qos(At-most-once))), 'Subscribe resend is as expected');
}

subtest 'Persistent reconnect', {
	my $topic = 'foo/bar';
	my $payload = 'Hello'.encode;
	my $qos = At-least-once;
	my $client5 = Protocol::MQTT::Client.new(:$client-identifier, :$keep-alive-interval, :$resend-interval, :persistent-session);
	$client5.connect;
	is($client5.state, Protocol::MQTT::Unconnected, 'We are connected');
	is-deeply($client5.next-events($now), [ Protocol::MQTT::Packet::Connect.new(:$client-identifier, :$keep-alive-interval) ], 'First packet is connect');
	$client5.received-packet(Protocol::MQTT::Packet::ConnAck.new, $now);
	is($client5.state, Protocol::MQTT::Connected, 'We are connected');
	$now++;
	my $supply = $client5.publish($topic, $payload, $qos, False, $now);
	my @messages = $client5.next-events($now);
	is(+@messages, 1, 'Sending one message');
	isa-ok(@messages[0], Protocol::MQTT::Packet::Publish, 'Message is a Publish');
	my $packet-id = @messages[0].packet-id;
	is-deeply(@messages[0], Protocol::MQTT::Packet::Publish.new(:$packet-id, :$topic, :$payload, :$qos, :!retain, :!dup), 'publish is as expected');

	$client5.connect;
	$now++;

	is($client5.state, Protocol::MQTT::Unconnected, 'We are connected');
	is-deeply($client5.next-events($now), [ Protocol::MQTT::Packet::Connect.new(:$client-identifier, :$keep-alive-interval) ], 'Next packet is second connect');
	$client5.received-packet(Protocol::MQTT::Packet::ConnAck.new(:session-acknowledge), $now);
	is($client5.state, Protocol::MQTT::Connected, 'We are connected');
	my @messages2 = $client5.next-events($now);
	is(+@messages2, 1, 'Sending one message again');
	is-deeply(@messages2[0], Protocol::MQTT::Packet::Publish.new(:$packet-id, :$topic, :$payload, :$qos, :!retain, :dup), 'Publish resend is as expected');
}

done-testing;


