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
|
/* This Source Code Form is subject to the terms of the Mozilla Public
* License, v. 2.0. If a copy of the MPL was not distributed with this
* file, You can obtain one at http://mozilla.org/MPL/2.0/. */
package org.mozilla.gecko.sync.repositories.uploaders;
import android.net.Uri;
import android.support.annotation.VisibleForTesting;
import org.mozilla.gecko.background.common.log.Logger;
import org.mozilla.gecko.sync.Server11PreviousPostFailedException;
import org.mozilla.gecko.sync.net.SyncStorageRequest;
import org.mozilla.gecko.sync.net.SyncStorageRequestDelegate;
import java.io.IOException;
import java.io.OutputStream;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import ch.boye.httpclientandroidlib.entity.ContentProducer;
import ch.boye.httpclientandroidlib.entity.EntityTemplate;
/**
* Responsible for creating and posting a <code>SyncStorageRequest</code> request object.
*/
public class RecordUploadRunnable implements Runnable {
public final String LOG_TAG = "RecordUploadRunnable";
public final static byte[] RECORDS_START = { 91 }; // [ in UTF-8
public final static byte[] RECORD_SEPARATOR = { 44 }; // , in UTF-8
public final static byte[] RECORDS_END = { 93 }; // ] in UTF-8
private static final String QUERY_PARAM_BATCH = "batch";
private static final String QUERY_PARAM_TRUE = "true";
private static final String QUERY_PARAM_BATCH_COMMIT = "commit";
private final MayUploadProvider mayUploadProvider;
private final SyncStorageRequestDelegate uploadDelegate;
private final ArrayList<byte[]> outgoing;
private final long byteCount;
// Used to construct POST URI during run().
@VisibleForTesting
public final boolean isCommit;
private final Uri collectionUri;
private final BatchMeta batchMeta;
public RecordUploadRunnable(MayUploadProvider mayUploadProvider,
Uri collectionUri,
BatchMeta batchMeta,
SyncStorageRequestDelegate uploadDelegate,
ArrayList<byte[]> outgoing,
long byteCount,
boolean isCommit) {
this.mayUploadProvider = mayUploadProvider;
this.uploadDelegate = uploadDelegate;
this.outgoing = outgoing;
this.byteCount = byteCount;
this.batchMeta = batchMeta;
this.collectionUri = collectionUri;
this.isCommit = isCommit;
}
public static class ByteArraysContentProducer implements ContentProducer {
ArrayList<byte[]> outgoing;
public ByteArraysContentProducer(ArrayList<byte[]> arrays) {
outgoing = arrays;
}
@Override
public void writeTo(OutputStream outstream) throws IOException {
int count = outgoing.size();
outstream.write(RECORDS_START);
if (count > 0) {
outstream.write(outgoing.get(0));
for (int i = 1; i < count; ++i) {
outstream.write(RECORD_SEPARATOR);
outstream.write(outgoing.get(i));
}
}
outstream.write(RECORDS_END);
}
public static long outgoingBytesCount(ArrayList<byte[]> outgoing) {
final long numberOfRecords = outgoing.size();
// Account for start and end tokens.
long count = RECORDS_START.length + RECORDS_END.length;
// Account for all the records.
for (int i = 0; i < numberOfRecords; i++) {
count += outgoing.get(i).length;
}
// Account for a separator between the records.
// There's one less separator than there are records.
if (numberOfRecords > 1) {
count += RECORD_SEPARATOR.length * (numberOfRecords - 1);
}
return count;
}
}
public static class ByteArraysEntity extends EntityTemplate {
private final long count;
public ByteArraysEntity(ArrayList<byte[]> arrays, long totalBytes) {
super(new ByteArraysContentProducer(arrays));
this.count = totalBytes;
this.setContentType("application/json");
// charset is set in BaseResource.
// Sanity check our byte counts.
long realByteCount = ByteArraysContentProducer.outgoingBytesCount(arrays);
if (realByteCount != totalBytes) {
throw new IllegalStateException("Mismatched byte counts. Received " + totalBytes + " while real byte count is " + realByteCount);
}
}
@Override
public long getContentLength() {
return count;
}
@Override
public boolean isRepeatable() {
return true;
}
}
@Override
public void run() {
if (!mayUploadProvider.mayUpload()) {
Logger.info(LOG_TAG, "Told not to proceed by the uploader. Cancelling upload, failing records.");
uploadDelegate.handleRequestError(new Server11PreviousPostFailedException());
return;
}
Logger.trace(LOG_TAG, "Running upload task. Outgoing records: " + outgoing.size());
// We don't want the task queue to proceed until this request completes.
// Fortunately, BaseResource is currently synchronous.
// If that ever changes, you'll need to block here.
final URI postURI = buildPostURI(isCommit, batchMeta, collectionUri);
final SyncStorageRequest request = new SyncStorageRequest(postURI);
request.delegate = uploadDelegate;
ByteArraysEntity body = new ByteArraysEntity(outgoing, byteCount);
request.post(body);
}
@VisibleForTesting
public static URI buildPostURI(boolean isCommit, BatchMeta batchMeta, Uri collectionUri) {
final Uri.Builder uriBuilder = collectionUri.buildUpon();
final String batchToken = batchMeta.getToken();
if (batchToken != null) {
uriBuilder.appendQueryParameter(QUERY_PARAM_BATCH, batchToken);
} else {
uriBuilder.appendQueryParameter(QUERY_PARAM_BATCH, QUERY_PARAM_TRUE);
}
if (isCommit) {
uriBuilder.appendQueryParameter(QUERY_PARAM_BATCH_COMMIT, QUERY_PARAM_TRUE);
}
try {
return new URI(uriBuilder.build().toString());
} catch (URISyntaxException e) {
throw new IllegalStateException("Failed to construct a collection URI", e);
}
}
}
|
