arzmq-sys 0.6.3

Low-level bindings to the zeromq library
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158

"""
pynorm - Python wrapper for NRL's libnorm
By: Tom Wambold <wambold@itd.nrl.navy.mil>
"""

# For Python 2.5
from __future__ import with_statement

from threading import Thread, Lock
from random import randint

#from .models import get_session, SessionProfile, SendProfile, ReceiveProfile
from .. import constants as c
from ..instance import Instance

class StopManager(Exception): pass

class Manager(Thread):
    """Provides a callback driven method of handling NORM events."""

    def __init__(self, instance=None, *args, **kwargs):
        super(Manager, self).__init__(*args, **kwargs)
        self.setDaemon(True)

        self.instance = instance if instance is not None else Instance()

        self.callbacks = {}
        self.sessions = {}
        self.lock = Lock()
        self.doQuit = False

    def run(self):
        while not self.doQuit:
            event = self.instance.getNextEvent(2)
            if event is None:
                continue

            with self.lock:
                try:
                    cbs = self.callbacks[event.type]
                except KeyError:
                    continue

                try:
                    for func, args, kwargs in cbs:
                        func(event, *args, **kwargs)
                except StopManager:
                    self.doQuit = True

    def register(self, event, func, *args, **kwargs):
        with self.lock:
            self.callbacks.setdefault(event, list()).append(
                    (func, args, kwargs))

    def unregister(self, event, func, *args, **kwargs):
        with self.lock:
            self.callbacks[event].remove((func, args, kwargs))

#    def start_session(self, name):
#        sqlsession = get_session()
#        profile = sqlsession.query(SessionProfile).filter_by(name=name).first()
#        self.instance.setCacheDirectory(profile.cache_directory)
#        session = self.instance.createSession(profile.address, profile.port, profile.node_id)
#        session.userData = name
#
#        if profile.interface is not None:
#            session.setMulticastInterface(profile.interface)
#
#        if profile.ttl is not None:
#            session.setTTL(profile.ttl)
#
#        if profile.tos is not None:
#            session.setTOS(profile.tos)
#
#        if profile.loopback is not None:
#            session.setLoopback(profile.loopback)
#        self.sessions[name] = session
#        sqlsession.close()
#
#    def start_sender(self, session_name, sender_name):
#        session = self.sessions[session_name]
#        sqlsession = get_session()
#        profile = sqlsession.query(SendProfile).filter_by(
#                name=sender_name).first()
#        session.startSender(randint(0, 10000), profile.buffer_space,
#                profile.segment_size, profile.block_size, profile.num_parity)
#
#        if profile.graceful_stop is not None:
#            session.sendGracefulStop = profile.graceful_stop
#
#        if profile.rate is not None:
#            session.setTxRate(profile.rate)
#
#        if profile.socket_buffer is not None:
#            session.setTxSocketBuffer(profile.socket_buffer)
#
#        if profile.congestion_control is not None:
#            session.setCongestionControl(profile.congestion_control)
#
#        if None not in (profile.rate_min, profile.rate_max):
#            session.setTxRateBounds(profile.rate_min, profile.rate_max)
#
#        if None not in (profile.size_max, profile.cache_min,
#                profile.cache_max):
#            session.setTxCacheBounds(profile.size_max, profile.cache_min,
#                    profile.cache_max)
#
#        if profile.auto_parity is not None:
#            session.setAutoParity(profile.auto_parity)
#
#        if profile.grtt is not None:
#            session.setGrttEstimate(profile.grtt)
#
#        if profile.grtt_max is not None:
#            session.setGrttMax(profile.grtt_max)
#
#        if profile.grtt_probing_mode is not None:
#            session.setGrttProbingMode(getattr(c, profile.grtt_probing_mode))
#
#        if None not in (profile.grtt_probing_interval_min,
#                profile.grtt_probing_interval_max):
#            session.setGrttProbingInterval(profile.grtt_probing_interval_min,
#                    profile.grtt_probing_interval_max)
#
#        if profile.backoff_factor is not None:
#            session.setBackoffFactor(profile.backoff_factor)
#
#        if profile.group_size is not None:
#            session.setGroupSize(profile.group_size)
#        sqlsession.close()
#        return session
#
#    def start_receiver(self, session_name, receiver_name):
#        sqlsession = get_session()
#        session = self.sessions[session_name]
#        profile = sqlsession.query(ReceiveProfile).filter_by(
#                name=receiver_name).first()
#        session.startReceiver(profile.buffer_space)
#
#        if profile.grace_period is not None:
#            session.gracePeriod = profile.grace_period
#
#        if profile.socket_buffer is not None:
#            session.setRxSocketBuffer(profile.socket_buffer)
#
#        if profile.is_silent is not None:
#            session.setSilentReceiver(profile.is_silent, profile.silent_max_delay)
#
#        if profile.unicast_nack is not None:
#            session.setDefaultUnicastNack(profile.unicast_nack)
#
#        if profile.nacking_mode is not None:
#            session.setDefaultNackingMode(getattr(c, profile.nacking_mode))
#
#        if profile.repair_boundary is not None:
#            session.setDefaultRepairBoundary(getattr(c, profile.repair_boundary))
#        sqlsession.close()
#        return session