9
8
from landscape.lib.fetch import fetch
10
from landscape.lib.fs import create_file
11
9
from landscape.lib import bpickle
12
10
from landscape.log import format_delta
13
11
from landscape import SERVER_API, VERSION
16
14
class HTTPTransport(object):
17
"""Transport makes a request to exchange message data over HTTP."""
19
def __init__(self, url, pubkey=None, payload_recorder=None):
21
@param url: URL of the remote Landscape server message system.
22
@param pubkey: SSH public key used for secure communication.
23
@param payload_recorder: PayloadRecorder used for recording exchanges
24
with the server. If C{None}, exchanges will not be recorded.
15
"""Transport makes a request to exchange message data over HTTP.
17
@param url: URL of the remote Landscape server message system.
18
@param pubkey: SSH public key used for secure communication.
21
def __init__(self, reactor, url, pubkey=None):
22
self._reactor = reactor
27
24
self._pubkey = pubkey
28
self._payload_recorder = payload_recorder
31
27
"""Get the URL of the remote message system."""
62
58
spayload = bpickle.dumps(payload)
63
if self._payload_recorder is not None:
64
self._payload_recorder.save(spayload)
66
60
start_time = time.time()
67
61
if logging.getLogger().getEffectiveLevel() <= logging.DEBUG:
68
62
logging.debug("Sending payload:\n%s", pprint.pformat(payload))
70
63
curly, data = self._curl(spayload, computer_id, message_api)
71
64
logging.info("Sent %d bytes and received %d bytes in %s.",
72
65
len(spayload), len(data),
95
class PayloadRecorder(object):
97
L{PayloadRecorder} records client exchanges with the server to disk for
100
Exchange payloads will be stored one per file, where the file name is
101
the elapsed time since the client was started.
104
def __init__(self, destination_dir):
106
@param destination_dir - The directory to record exchanges in.
108
self._time_offset = time.time()
109
self._destination_dir = destination_dir
110
self._last_payload_time = -1
111
if self._destination_dir is not None:
112
self._create_destination_dir(self._destination_dir)
113
self._delete_old_payloads()
115
def _create_destination_dir(self, destination_dir):
116
"""Create the destination directory if it does not exist.
118
@param destination_dir: The directory to be created.
120
if not os.path.exists(destination_dir):
121
os.mkdir(destination_dir)
123
def _delete_old_payloads(self):
124
"""Delete payloads lying around from a previous session."""
125
for filename in os.listdir(self._destination_dir):
126
file_path = os.path.join(self._destination_dir, filename)
127
if os.path.isfile(file_path):
130
def save(self, payload):
131
"""Persist the given payload to disk.
133
@param payload: The payload to be persisted.
135
payload_name = self.get_payload_filename()
136
create_file(os.path.join(self._destination_dir, payload_name),
139
def get_payload_filename(self):
141
Generate a payload filename. This method ensures that payloads
142
will have a unique name.
144
payload_time = time.time() - self._time_offset
145
last_payload_time = '%.3f' % self._last_payload_time
146
this_payload_time = '%.3f' % payload_time
147
if last_payload_time == this_payload_time:
148
payload_time = payload_time + 0.001
149
self._last_payload_time = payload_time
150
return '%.3f' % payload_time
153
88
class FakeTransport(object):
154
89
"""Fake transport for testing purposes."""
156
def __init__(self, url=None, pubkey=None, payload_recorder=None):
91
def __init__(self, reactor=None, url=None, pubkey=None):
157
92
self._pubkey = pubkey
158
self._payload_recorder = payload_recorder
159
93
self.payloads = []
160
94
self.responses = []
161
95
self._current_response = 0
added
removed
landscape/broker/transport.py
