66 lines
2.0 KiB
Python
66 lines
2.0 KiB
Python
import base64
|
|
import random
|
|
from collections import Counter
|
|
from enum import Enum
|
|
|
|
from colony.core.roles.role import Role, Command
|
|
from colony.core.storages.replica import ReplicaStorage
|
|
|
|
|
|
class Replicator(Role):
|
|
class COMMANDS(Enum):
|
|
# Write to replicator
|
|
REPLICATE = 'Replicator.Replicate'
|
|
# Read from replicator
|
|
RETRIEVE = 'Replicator.Retrieve'
|
|
HASH_SUM = 'Replicator.HashSum'
|
|
|
|
def __init__(self, context, storage=None):
|
|
commands = {
|
|
self.COMMANDS.REPLICATE.value: Command(self.replicate),
|
|
self.COMMANDS.RETRIEVE.value: Command(self.retrieve),
|
|
self.COMMANDS.HASH_SUM.value: Command(self.hash_sum),
|
|
}
|
|
|
|
self.storage = storage or ReplicaStorage()
|
|
|
|
super().__init__(commands, context)
|
|
|
|
def start(self):
|
|
replicator_checksums = dict()
|
|
|
|
for node in self.context.nodes_with_role(Replicator):
|
|
if (node.ip, node.port) == self.context.membership().key:
|
|
continue
|
|
|
|
response = self.context.execute_on(node, Replicator.COMMANDS.HASH_SUM)
|
|
if response['status'] != 'ok':
|
|
continue
|
|
|
|
checksum = response['result']
|
|
|
|
replicator_checksums[node] = checksum
|
|
|
|
if replicator_checksums:
|
|
checksum = Counter(replicator_checksums.values()).most_common(1)[0][0]
|
|
matching_nodes = [
|
|
node
|
|
for node, node_checksum in replicator_checksums.items()
|
|
if node_checksum == checksum
|
|
]
|
|
|
|
if checksum != self.hash_sum():
|
|
node = random.choice(matching_nodes)
|
|
response = self.context.execute_on(node, Replicator.COMMANDS.RETRIEVE)
|
|
self.storage.restore(base64.b64decode(response['result']))
|
|
|
|
def replicate(self, filename, content):
|
|
self.storage.save(filename, content)
|
|
return True
|
|
|
|
def retrieve(self):
|
|
return base64.b64encode(self.storage.archive()).decode()
|
|
|
|
def hash_sum(self):
|
|
return self.storage.dir_checksum()
|