Source code for abce.postbox

from multiprocessing.managers import BaseManager
from collections import defaultdict
from copy import deepcopy


[docs]class Postbox:
    def __init__(self):
        self.boxes = defaultdict(dict)

[docs]    def receive(self, group, id):
        return self.boxes[group][id]

[docs]    def clear(self, group):
        self.boxes[group] = {}

[docs]    def copy(self, group_names):
        return {group: deepcopy(self.boxes[group]) for group in group_names}

[docs]    def send(self, group, id, message):
        try:
            self.boxes[group][id].append(message)
        except KeyError:
            self.boxes[group][id] = [message]


[docs]class PostboxManager(BaseManager):
    pass