← Back to branch summary

~ubuntuone-control-tower/ubuntuone-storage-protocol/stable-1-6

« back to all changes in this revision

Viewing changes to ubuntuone/storageprotocol/client.py

  • Committer: Tarmac
  • Author(s): Rodney Dawes
  • Date: 2010-11-16 15:10:48 UTC
  • mfrom: (121.1.4 use-packages)
  • Revision ID: tarmac-20101116151048-b0e20j7lorb4yhe1
Switch to using packaged mocker and ubuntuone-dev-tools
Use pyflakes with u1lint and also run pep8
Fix a lot of pylint/pyflakes/pep8 errors

Show diffs side-by-side

added added

removed removed

Lines of Context:
ubuntuone/storageprotocol/client.py
30
30
from twisted.python import log
31
31
 
32
32
from ubuntuone.storageprotocol import (
33
 
    protocol_pb2, request, sharersp, volumes, delta
34
 
)
 
33
    protocol_pb2, request, sharersp, volumes, delta)
35
34
 
36
35
log_debug = partial(log.msg, loglevel=logging.DEBUG)
37
36
 
 
37
 
38
38
class StorageClient(request.RequestHandler):
39
39
    """A Basic Storage Protocol client."""
40
40
 
83
83
            except ValueError:
84
84
                return
85
85
            self.line_mode = False
86
 
            data = data[pos+2:]
 
86
            data = data[pos + 2:]
87
87
 
88
88
        request.RequestHandler.dataReceived(self, data)
89
89
 
110
110
        object when completed.
111
111
 
112
112
        """
113
 
        request = oauth.OAuthRequest.from_consumer_and_token(
 
113
        req = oauth.OAuthRequest.from_consumer_and_token(
114
114
            oauth_consumer=consumer,
115
115
            token=token,
116
116
            http_method="GET",
117
117
            http_url="storage://server")
118
 
        request.sign_request(
 
118
        req.sign_request(
119
119
            oauth.OAuthSignatureMethod_PLAINTEXT(), consumer, token)
120
120
 
121
121
        # Make sure all parameter values are strings.
122
122
        auth_parameters = dict(
123
 
            (key, str(value)) for key, value in request.parameters.iteritems())
 
123
            (key, str(value)) for key, value in req.parameters.iteritems())
124
124
        p = Authenticate(self, auth_parameters)
125
125
        p.start()
126
126
        return p.deferred
382
382
        else:
383
383
            raise TypeError('callback for share_change must be callable')
384
384
 
385
 
 
386
385
    def notify_share_change(self, notify_share):
387
386
        """Call the current changed share callback, if any, with the
388
387
        notify info.
418
417
        else:
419
418
            raise TypeError('callback for share_answer must be callable')
420
419
 
421
 
 
422
420
    def notify_share_answer(self, msg):
423
421
        """Call the current share answer callback, if any, with the info.
424
422
 
595
593
 
596
594
    def processMessage(self, message):
597
595
        """Process messages."""
598
 
        # pylint: disable-msg=W0201
599
596
        # pylint: disable=W0201
600
597
        if message.type == protocol_pb2.Message.NODE_ATTR:
601
598
            if self.node_attr_callback is not None:
645
642
 
646
643
    def _start(self):
647
644
        """Send the LIST_SHARES message to the server."""
648
 
        # pylint: disable-msg=W0201
649
645
        # pylint: disable=W0201
650
646
        message = protocol_pb2.Message()
651
647
        message.type = protocol_pb2.Message.LIST_SHARES
903
899
 
904
900
    @ivar new_generation: the generation that the volume is at now
905
901
    """
906
 
    # pylint: disable-msg=C0111
907
902
    # pylint: disable=C0111
908
903
 
909
904
    def __init__(self, protocol, share, node_id):
1035
1030
        """
1036
1031
        request.Request.__init__(self, protocol)
1037
1032
        self.query_message = qm = protocol_pb2.Message()
1038
 
        qm.id = 0 # just to have something in the field when calculating size
 
1033
        qm.id = 0  # just to have something in the field when calculating size
1039
1034
        qm.type = protocol_pb2.Message.QUERY
1040
1035
        self.response = []
1041
1036
        self.overflow = None
1078
1073
class BytesMessageProducer(object):
1079
1074
    """Produce BYTES messages from a file."""
1080
1075
 
1081
 
    def __init__(self, request, fh, offset):
 
1076
    def __init__(self, req, fh, offset):
1082
1077
        """Create a BytesMessageProducer."""
1083
 
        self.request = request
 
1078
        self.request = req
1084
1079
        self.producing = False
1085
1080
        self.fh = fh
1086
1081
        self.offset = offset
1203
1198
    node id of the created object.
1204
1199
 
1205
1200
    @cvar create_message: must be overridden with the correct creation message
1206
 
        to send
1207
 
    @cvar response_message: must be overridden with the correct creation success
1208
 
        message that will be received
 
1201
    to send
 
1202
    @cvar response_message: must be overridden with the correct creation
 
1203
    success message that will be received
1209
1204
 
1210
1205
    @ivar new_id: the id of the node that was created (available upon success)
1211
1206
    @ivar new_parent_id: the parent id the node now exists under
1240
1235
 
1241
1236
    def processMessage(self, message):
1242
1237
        """Handle messages."""
1243
 
        # pylint: disable-msg=W0201
1244
1238
        # pylint: disable=W0201
1245
1239
        if message.type == self.create_response:
1246
1240
            self.new_id = message.new.node
1286
1280
 
1287
1281
    def processMessage(self, message):
1288
1282
        """Handle messages."""
1289
 
        # pylint: disable-msg=W0201
1290
1283
        # pylint: disable=W0201
1291
1284
        if message.type == protocol_pb2.Message.PROTOCOL_VERSION:
1292
1285
            self.other_protocol_version = message.protocol.version
1387
1380
 
1388
1381
    def _start(self):
1389
1382
        """Send the FREE_SPACE_INQUIRY message to the server."""
1390
 
        # pylint: disable-msg=W0201
1391
1383
        # pylint: disable=W0201
1392
1384
        message = protocol_pb2.Message()
1393
1385
        message.type = protocol_pb2.Message.FREE_SPACE_INQUIRY
1397
1389
 
1398
1390
    def processMessage(self, message):
1399
1391
        """Process the answer from the server."""
1400
 
        # pylint: disable-msg=W0201
1401
1392
        # pylint: disable=W0201
1402
1393
        if message.type == protocol_pb2.Message.FREE_SPACE_INFO:
1403
1394
            self.free_bytes = message.free_space_info.free_bytes
1411
1402
 
1412
1403
    def _start(self):
1413
1404
        """Send the FREE_SPACE_INQUIRY message to the server."""
1414
 
        # pylint: disable-msg=W0201
1415
1405
        # pylint: disable=W0201
1416
1406
        message = protocol_pb2.Message()
1417
1407
        message.type = protocol_pb2.Message.ACCOUNT_INQUIRY
1420
1410
 
1421
1411
    def processMessage(self, message):
1422
1412
        """Process the answer from the server."""
1423
 
        # pylint: disable-msg=W0201
1424
1413
        # pylint: disable=W0201
1425
1414
        if message.type == protocol_pb2.Message.ACCOUNT_INFO:
1426
1415
            self.purchased_bytes = message.account_info.purchased_bytes
1428
1417
        else:
1429
1418
            self._default_process_message(message)
1430
1419
 
 
1420
 
1431
1421
class GetDelta(request.Request):
1432
1422
    """Get a delta on a volume
1433
1423
 
1491
1481
        else:
1492
1482
            self._default_process_message(message)
1493
1483
 
 
1484
 
1494
1485
class ThrottlingStorageClient(StorageClient):
1495
1486
    """The throttling version of the StorageClient protocol."""
1496
1487
 
1545
1536
 
1546
1537
class StorageClientFactory(ClientFactory):
1547
1538
    """StorageClient factory."""
1548
 
    # pylint: disable-msg=W0232
1549
1539
    # pylint: disable=W0232
1550
1540
    protocol = StorageClient
1551
1541
 
1559
1549
    def __init__(self, throttling_enabled=False,
1560
1550
                 read_limit=None, write_limit=None):
1561
1551
        """Create the instance."""
1562
 
        self._readLimit = None # max bytes we should read per second
1563
 
        self._writeLimit = None # max bytes we should write per second
 
1552
        self._readLimit = None  # max bytes we should read per second
 
1553
        self._writeLimit = None  # max bytes we should write per second
1564
1554
        self._throttling_reads = False
1565
1555
        self._throttling_writes = False
1566
1556
        self._set_read_limit(read_limit)
1599
1589
        if not self.valid_limit(limit):
1600
1590
            raise ValueError('Read limit must be greater than 0.')
1601
1591
        self._readLimit = limit
1602
 
    # it's a property, pylint: disable-msg=W0212
1603
1592
    # it's a property, pylint: disable=W0212
1604
1593
    readLimit = property(lambda self: self._readLimit, _set_read_limit)
1605
1594
    writeLimit = property(lambda self: self._writeLimit, _set_write_limit)
1606
 
    # pylint: enable-msg=W0212
1607
1595
    # pylint: enable=W0212
1608
1596
 
1609
1597
    def callLater(self, period, func, *args, **kwargs):
1610
1598
        """Wrapper around L{reactor.callLater} for test purpose."""
1611
1599
        return reactor.callLater(period, func, *args, **kwargs)
1612
1600
 
1613
 
    def maybeCallLater(self, id, period, func):
 
1601
    def maybeCallLater(self, call_id, period, func):
1614
1602
        """Maybe run callLater(period, func).
1615
1603
 
1616
1604
        Only if we don't have a DelayedCall with the
1617
1605
        specified id already running.
1618
1606
        """
1619
 
        delayed_call = getattr(self, id)
 
1607
        delayed_call = getattr(self, call_id)
1620
1608
        # check if we already have a DelayedCall running
1621
1609
        if delayed_call is None or (not delayed_call.active() \
1622
1610
           and delayed_call.cancelled):
1635
1623
            self.readThisSecond += length
1636
1624
            self.checkReadBandwidth()
1637
1625
 
1638
 
    def _get_throttle_time(self, bytes, limit):
1639
 
        """Calculate the throttle_time for bytes and limit."""
1640
 
        return (float(bytes) / limit) - 1.0
 
1626
    def _get_throttle_time(self, data_bytes, limit):
 
1627
        """Calculate the throttle_time for data_bytes and limit."""
 
1628
        return (float(data_bytes) / limit) - 1.0
1641
1629
 
1642
1630
    def checkReadBandwidth(self):
1643
1631
        """Check if we've passed bandwidth limits."""
1676
1664
        self.resetReadThisSecondID = self.callLater(1,
1677
1665
                                                   self._resetReadThisSecond)
1678
1666
 
1679
 
    def _resetWrittenThisSecond (self):
 
1667
    def _resetWrittenThisSecond(self):
1680
1668
        """Reset the counter named with 'name' every 1 second."""
1681
1669
        self.writtenThisSecond = 0
1682
 
        self.resetWriteThisSecondID = self.callLater(1,
1683
 
                                                self._resetWrittenThisSecond)
 
1670
        self.resetWriteThisSecondID = self.callLater(
 
1671
            1, self._resetWrittenThisSecond)
1684
1672
 
1685
1673
    def throttleReads(self):
1686
1674
        """Throttle reads on all protocols."""