From 6de217a02fcf206eec209ef98f41dbd5789ef7f5 Mon Sep 17 00:00:00 2001 From: Michael Pivato Date: Fri, 18 Jan 2019 16:40:51 +1030 Subject: [PATCH] Add callbacks, debug print lines --- .../CentralisedDecision/commander.py | 51 ++++++++++++------- 1 file changed, 33 insertions(+), 18 deletions(-) diff --git a/DecisionSystem/CentralisedDecision/commander.py b/DecisionSystem/CentralisedDecision/commander.py index 490dce7..fcb7973 100644 --- a/DecisionSystem/CentralisedDecision/commander.py +++ b/DecisionSystem/CentralisedDecision/commander.py @@ -1,6 +1,6 @@ import paho.mqtt.client as mqtt import time -from DecisionSystem.messages import Message, CommanderWill, RequestVote +from DecisionSystem.messages import Message, CommanderWill, RequestVote, GetSwarmParticipants, deserialise import json import numpy as np @@ -26,28 +26,28 @@ class Commander: self._cfg = cfg self.client = mqtt.Client() - self.client.connect(cfg["mqtt"]["host"], cfg["mqtt"]["port"], cfg["mqtt"]["host"]) - - self.client.subscribe("swarm1/commander") - # Commander needs a will message too, for the decentralised version, so the - # voters know to pick a new commander. - # If using apache zookeeper this won't be needed. - self.client.publish("swarm1/voters", CommanderWill(self.client._client_id)) + self.client.on_connect = self.on_connect + self.client.on_message = self.on_message + self.client.on_disconnect = self.on_disconnect + self.client.connect(cfg["mqtt"]["host"], int(cfg["mqtt"]["port"]), int(cfg["mqtt"]["timeout"])) + # Don't forget to start the event loop!!! + self.client.loop_start() def make_decision(self): # Should change this to follow strategy pattern, for different implementations of # making a decision on the votes. + print("Making a decision") votes = self._votes dif_votes = {} for vote in votes: # Get the count of different votes. - if str(vote) in dif_votes: - dif_votes[str(vote)] = dif_votes[str(vote)] + 1 + if vote in dif_votes: + dif_votes[vote] = dif_votes[vote] + 1 else: - dif_votes[str(vote)] = 1 + dif_votes[vote] = 1 - max_vote = None + max_vote = "" max_vote_num = 0 # Should try using a numpy array for this. @@ -56,33 +56,40 @@ class Commander: max_vote = vote max_vote_num = dif_votes[vote] + print("Made Decision!") return max_vote def get_votes(self): - # Should consider abstracting these messages to another class for portability. + # Should abstract messaging to another class. + print("Gathering Votes") self._taking_votes = True # Publish a message that votes are needed. - self.client.publish("swarm1/voters", RequestVote(self.client._client_id)) + print("Sending request message") + self.client.publish("swarm1/voters", RequestVote(self.client._client_id).serialise()) + print("published message") time.sleep(self.timeout) self._taking_votes = False - self.make_decision() + return self.make_decision() def on_message(self, client, userdata, message): + print("Message Received") messageD = None try: - messageD = Message().deserialise(message.payload) + messageD = deserialise(message.payload) except: print("Incorrect Message Has Been Sent") return - # Need to consider that a malicious message may have a type with incorrect subtypes. if messageD.type == "connect": + print("Voter connected!") # Voter just connected/reconnnected. if not messageD["client"] in self._connectedVoters: self._connectedVoters.append(messageD["client"]) elif messageD.type == "vote": + print("Received a vote!") # Voter is sending in their vote. + print(messageD.data["vote"]) if self._taking_votes: # Commander must have requested their taking votes, and the timeout # has not occurred. @@ -91,11 +98,19 @@ class Commander: self._votes[messageD.sender] = messageD.data["vote"] elif messageD.type == "disconnected": + print("Voter disconnected :(") self._connectedVoters.remove(messageD.sender) def on_connect(self, client, userdata, flags, rc): self.client.subscribe("swarm1/commander") def get_participants(self): + self.client.publish("swarm1/voters", GetSwarmParticipants().serialise()) + # Commander needs a will message too, for the decentralised version, so the + # voters know to pick a new commander. + # If using apache zookeeper this won't be needed. + # That's the wrong method for setting a will. + # self.client.publish("swarm1/voters", CommanderWill(self.client._client_id).serialise()) - self.client.publish("swarm1/voters") \ No newline at end of file + def on_disconnect(self, client, userdata, rc): + pass \ No newline at end of file