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
|
# -*- coding: utf-8 -*-
# send.py
# Copyright (C) 2015 LEAP
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import json
from twisted.internet import defer
from leap.soledad.common.log import getLogger
from leap.soledad.client.events import emit_async
from leap.soledad.client.events import SOLEDAD_SYNC_SEND_STATUS
from leap.soledad.client.http_target.support import RequestBody
from .send_protocol import DocStreamProducer
logger = getLogger(__name__)
class HTTPDocSender(object):
"""
Handles Document uploading from Soledad server, using HTTP as transport.
They need to be encrypted and metadata prepared before sending.
"""
MAX_BATCH_SIZE = 0 # disabled by now, this is being tested yet
# The uuid of the local replica.
# Any class inheriting from this one should provide a meaningful attribute
# if the sync status event is meant to be used somewhere else.
uuid = 'undefined'
userid = 'undefined'
@defer.inlineCallbacks
def _send_docs(self, docs_by_generation, last_known_generation,
last_known_trans_id, sync_id):
if not docs_by_generation:
defer.returnValue([None, None])
# add remote replica metadata to the request
body = RequestBody(
last_known_generation=last_known_generation,
last_known_trans_id=last_known_trans_id,
sync_id=sync_id,
ensure=self._ensure_callback is not None)
result = yield self._send_batch(body, docs_by_generation)
response_dict = json.loads(result)[0]
gen_after_send = response_dict['new_generation']
trans_id_after_send = response_dict['new_transaction_id']
defer.returnValue([gen_after_send, trans_id_after_send])
@defer.inlineCallbacks
def _send_batch(self, body, docs):
total = len(docs)
missing = total - body.consumed
calls = []
for i in xrange(1, missing + 1):
idx = body.consumed + i
entry = docs[idx - 1]
calls.append((self._prepare_one_doc,
entry, body, idx, total))
result = yield self._send_request(body, calls)
_emit_send_status(self.uuid, body.consumed, total)
defer.returnValue(result)
def _send_request(self, body, calls):
return self._http_request(
self._url,
method='POST',
body=(body, calls),
content_type='application/x-soledad-sync-put',
body_producer=DocStreamProducer)
@defer.inlineCallbacks
def _prepare_one_doc(self, entry, body, idx, total):
get_doc, gen, trans_id = entry
doc, content = yield self._encrypt_doc(get_doc)
body.insert_info(
id=doc.doc_id, rev=doc.rev, content=content, gen=gen,
trans_id=trans_id, number_of_docs=total,
doc_idx=idx)
@defer.inlineCallbacks
def _encrypt_doc(self, get_doc):
if type(get_doc) == tuple:
f, args = get_doc
doc = yield f(args)
else:
# tests
doc = get_doc
if doc.is_tombstone():
defer.returnValue((doc, None))
else:
# TODO -- for blobs, should stream the doc raw content
# TODO -- get rid of this json encoding
content = yield self._crypto.encrypt_doc(doc)
defer.returnValue((doc, content))
def _emit_send_status(user_data, idx, total):
content = {'sent': idx, 'total': total}
emit_async(SOLEDAD_SYNC_SEND_STATUS, user_data, content)
msg = "%d/%d" % (idx, total)
logger.debug("Sync send status: %s" % msg)
|
