Skip to main content
aboutsummaryrefslogtreecommitdiffstats
blob: 03122b1b3ca203d2bb80b0c72f8d6dbf52958657 (plain) (blame)
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
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
# *****************************************************************************
# * Copyright (c) 2011, 2013 Wind River Systems, Inc. and others.
# * All rights reserved. This program and the accompanying materials
# * are made available under the terms of the Eclipse Public License v1.0
# * which accompanies this distribution, and is available at
# * http://www.eclipse.org/legal/epl-v10.html
# *
# * Contributors:
# *     Wind River Systems - initial API and implementation
# *****************************************************************************

"""
Both hosts and targets are represented by objects
implementing IPeer interface. A peer can act as host or
target depending on services it implements.
List of currently known peers can be retrieved by
calling Locator.getPeers()

A TCF agent houses one or more service managers. A service manager has a one or
more services to expose to the world. The service manager creates one or more
peers to represent itself, one for every access path the agent is
reachable by. For example, in agents accessible via TCP/IP, the
service manger would create a peer for every subnet it wants to participate in.
All peers of particular service manager represent identical sets of services.
"""

import os
import time
from . import protocol, transport, services, channel
from .services import locator

# Peer unique ID
ATTR_ID = "ID"

# Unique ID of service manager that is represented by this peer
ATTR_SERVICE_MANAGER_ID = "ServiceManagerID"

# Agent unique ID
ATTR_AGENT_ID = "AgentID"

# Peer name
ATTR_NAME = "Name"

# Name of the peer operating system
ATTR_OS_NAME = "OSName"

# Transport name, for example TCP, SSL
ATTR_TRANSPORT_NAME = "TransportName"

# If present, indicates that the peer can forward traffic to other peers
ATTR_PROXY = "Proxy"

# Host DNS name or IP address
ATTR_IP_HOST = "Host"

# Optional list of host aliases
ATTR_IP_ALIASES = "Aliases"

# Optional list of host addresses
ATTR_IP_ADDRESSES = "Addresses"

# IP port number, must be decimal number
ATTR_IP_PORT = "Port"


class Peer(object):
    def __init__(self, attrs):
        self.attrs = attrs

    def getAttributes(self):
        """@return map of peer attributes"""
        return self.attrs

    def getID(self):
        """@return peer unique ID, same as getAttributes().get(ATTR_ID)"""
        return self.attrs.get(ATTR_ID)

    def getServiceManagerID(self):
        """@return service manager unique ID, same as
        getAttributes().get(ATTR_SERVICE_MANAGER_ID)"""
        assert protocol.isDispatchThread()
        return self.attrs.get(ATTR_SERVICE_MANAGER_ID)

    def getAgentID(self):
        """@return agent unique ID, same as
        getAttributes().get(ATTR_AGENT_ID)"""
        assert protocol.isDispatchThread()
        return self.attrs.get(ATTR_AGENT_ID)

    def getName(self):
        """@return peer name, same as getAttributes().get(ATTR_NAME)"""
        return self.attrs.get(ATTR_NAME)

    def getOSName(self):
        """@return agent OS name, same as getAttributes().get(ATTR_OS_NAME)"""
        return self.attrs.get(ATTR_OS_NAME)

    def getTransportName(self):
        """@return transport name, same as
        getAttributes().get(ATTR_TRANSPORT_NAME)"""
        return self.attrs.get(ATTR_TRANSPORT_NAME)

    def openChannel(self):
        """Open channel to communicate with this peer.
        Note: the channel is not fully open yet when this method returns.
        Its state is channel.STATE_OPENING.
        Protocol.ChannelOpenListener and IChannel.IChannelListener listeners
        will be called when the channel will change state to open or closed.
        Clients are supposed to register IChannel.IChannelListener right after
        calling openChannel(), or, at least, in same dispatch cycle.
        For example:
                 channel = peer.openChannel()
                 channel.addChannelListener(...)
        """
        raise RuntimeError("Abstract method")


class TransientPeer(Peer):
    """
    Transient implementation of IPeer interface.
    Objects of this class are not tracked by Locator service.
    See AbstractPeer for IPeer objects that should go into the Locator table.
    """
    def __init__(self, attrs):
        self.rw_attrs = {}
        self.rw_attrs.update(attrs)
        # TODO readonly map
        ro_attrs = {}
        ro_attrs.update(self.rw_attrs)
        super(TransientPeer, self).__init__(ro_attrs)

    def openChannel(self):
        return transport.openChannel(self)


class LocalPeer(TransientPeer):
    """
    LocalPeer object represents local end-point of TCF communication channel.
    There should be exactly one such object in a TCF agent.
    The object can be used to open a loop-back communication channel that
    allows the agent to access its own services same way as remote services.
    Note that "local" here is relative to the agent, and not same as in
    "local host".
    """
    def __init__(self):
        super(LocalPeer, self).__init__(self.createAttributes())

    def createAttributes(self):
        attrs = {
            ATTR_ID: "TCFLocal",
            ATTR_SERVICE_MANAGER_ID: services.getServiceManagerID(),
            ATTR_AGENT_ID: protocol.getAgentID(),
            ATTR_NAME: "Local Peer",
            ATTR_OS_NAME: os.name,
            ATTR_TRANSPORT_NAME: "Loop"
        }
        return attrs


class AbstractPeer(TransientPeer):
    """
    Abstract implementation of IPeer interface.
    Objects of this class are stored in Locator service peer table.
    The class implements sending notification events to Locator listeners.
    See TransientPeer for IPeer objects that are not stored in the Locator
    table.
    """

    def __init__(self, attrs):
        super(AbstractPeer, self).__init__(attrs)
        assert protocol.isDispatchThread()
        self.last_heart_beat_time = 0
        _id = self.getID()
        assert _id
        peers = protocol.getLocator().getPeers()
        if isinstance(peers.get(_id), RemotePeer):
            peers.get(_id).dispose()
        assert _id not in peers
        peers[_id] = self
        self.sendPeerAddedEvent()

    def dispose(self):
        assert protocol.isDispatchThread()
        _id = self.getID()
        assert _id
        peers = protocol.getLocator().getPeers()
        assert peers.get(_id) == self
        del peers[_id]
        self.sendPeerRemovedEvent()

    def onChannelTerminated(self):
        # A channel to this peer was terminated:
        # not delaying next heart beat helps client to recover much faster.
        self.last_heart_beat_time = 0

    def updateAttributes(self, attrs):
        equ = True
        assert attrs.get(ATTR_ID) == self.rw_attrs.get(ATTR_ID)
        for key in self.rw_attrs.keys():
            if self.rw_attrs.get(key) != attrs.get(key):
                equ = False
                break
        for key in attrs.keys():
            if attrs.get(key) != self.rw_attrs.get(key):
                equ = False
                break
        timeVal = int(time.time() * 1000)
        if not equ:
            self.rw_attrs.clear()
            self.rw_attrs.update(attrs)
            for l in protocol.getLocator().getListeners():
                try:
                    l.peerChanged(self)
                except Exception as x:
                    protocol.log("Unhandled exception in Locator listener", x)
            try:
                args = [self.rw_attrs]
                protocol.sendEvent(locator.NAME, "peerChanged",
                                   channel.toJSONSequence(args))
            except IOError as x:
                protocol.log("Locator: failed to send 'peerChanged' event", x)
            self.last_heart_beat_time = timeVal
        elif self.last_heart_beat_time + locator.DATA_RETENTION_PERIOD / 4 \
                                                                    < timeVal:
            for l in protocol.getLocator().getListeners():
                try:
                    l.peerHeartBeat(attrs.get(ATTR_ID))
                except Exception as x:
                    protocol.log("Unhandled exception in Locator listener", x)
            try:
                args = [self.rw_attrs.get(ATTR_ID)]
                protocol.sendEvent(locator.NAME, "peerHeartBeat",
                                   channel.toJSONSequence(args))
            except IOError as x:
                protocol.log(
                    "Locator: failed to send 'peerHeartBeat' event", x)
            self.last_heart_beat_time = timeVal

    def sendPeerAddedEvent(self):
        for l in protocol.getLocator().getListeners():
            try:
                l.peerAdded(self)
            except Exception as x:
                protocol.log("Unhandled exception in Locator listener", x)
        try:
            args = [self.rw_attrs]
            protocol.sendEvent(locator.NAME, "peerAdded",
                               channel.toJSONSequence(args))
        except IOError as x:
            protocol.log("Locator: failed to send 'peerAdded' event", x)
        self.last_heart_beat_time = int(time.time() * 1000)

    def sendPeerRemovedEvent(self):
        for l in protocol.getLocator().getListeners():
            try:
                l.peerRemoved(self.rw_attrs.get(ATTR_ID))
            except Exception as x:
                protocol.log("Unhandled exception in Locator listener", x)
        try:
            args = [self.rw_attrs.get(ATTR_ID)]
            protocol.sendEvent(locator.NAME, "peerRemoved",
                               channel.toJSONSequence(args))
        except IOError as x:
            protocol.log("Locator: failed to send 'peerRemoved' event", x)


class RemotePeer(AbstractPeer):
    """
    RemotePeer objects represent TCF agents that Locator service discovered on
    local network.
    This includes both local host agents and remote host agents.
    Note that "remote peer" means any peer accessible over network,
    it does not imply the agent is running on a "remote host".
    If an agent binds multiple network interfaces or multiple ports, it can be
    represented by multiple RemotePeer objects - one per each network
    address/port combination.
    RemotePeer objects life cycle is managed by Locator service.
    """

    def __init__(self, attrs):
        super(RemotePeer, self).__init__(attrs)
        self.last_update_time = int(time.time() * 1000)

    def updateAttributes(self, attrs):
        super(RemotePeer, self).updateAttributes(attrs)
        self.last_update_time = int(time.time() * 1000)

    def getLastUpdateTime(self):
        return self.last_update_time

Back to the top