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 package org.eclipse.jgit.internal.ketch;
45
46 import static org.eclipse.jgit.internal.ketch.KetchReplica.CommitMethod.ALL_REFS;
47 import static org.eclipse.jgit.lib.Ref.Storage.NETWORK;
48 import static org.eclipse.jgit.transport.ReceiveCommand.Result.LOCK_FAILURE;
49 import static org.eclipse.jgit.transport.ReceiveCommand.Result.NOT_ATTEMPTED;
50 import static org.eclipse.jgit.transport.ReceiveCommand.Result.OK;
51 import static org.eclipse.jgit.transport.ReceiveCommand.Result.REJECTED_NODELETE;
52 import static org.eclipse.jgit.transport.ReceiveCommand.Result.REJECTED_NONFASTFORWARD;
53 import static org.eclipse.jgit.transport.ReceiveCommand.Result.REJECTED_OTHER_REASON;
54
55 import java.io.IOException;
56 import java.util.ArrayList;
57 import java.util.Collection;
58 import java.util.Collections;
59 import java.util.LinkedHashMap;
60 import java.util.List;
61 import java.util.Map;
62
63 import org.eclipse.jgit.annotations.Nullable;
64 import org.eclipse.jgit.errors.NotSupportedException;
65 import org.eclipse.jgit.errors.TransportException;
66 import org.eclipse.jgit.lib.AnyObjectId;
67 import org.eclipse.jgit.lib.NullProgressMonitor;
68 import org.eclipse.jgit.lib.ObjectId;
69 import org.eclipse.jgit.lib.ObjectIdRef;
70 import org.eclipse.jgit.lib.Ref;
71 import org.eclipse.jgit.lib.Repository;
72 import org.eclipse.jgit.transport.FetchConnection;
73 import org.eclipse.jgit.transport.PushConnection;
74 import org.eclipse.jgit.transport.ReceiveCommand;
75 import org.eclipse.jgit.transport.RemoteConfig;
76 import org.eclipse.jgit.transport.RemoteRefUpdate;
77 import org.eclipse.jgit.transport.Transport;
78 import org.eclipse.jgit.transport.URIish;
79
80
81
82
83
84
85
86
87
88 public class RemoteGitReplica extends KetchReplica {
89 private final URIish uri;
90 private final RemoteConfig remoteConfig;
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107 public RemoteGitReplica(KetchLeader leader, String name, URIish uri,
108 ReplicaConfig cfg, @Nullable RemoteConfig rc) {
109 super(leader, name, cfg);
110 this.uri = uri;
111 this.remoteConfig = rc;
112 }
113
114
115
116
117
118
119 public URIish getURI() {
120 return uri;
121 }
122
123
124
125
126
127
128 @Nullable
129 protected RemoteConfig getRemoteConfig() {
130 return remoteConfig;
131 }
132
133
134 @Override
135 protected String describeForLog() {
136 return String.format("%s @ %s", getName(), getURI());
137 }
138
139
140 @Override
141 protected void startPush(ReplicaPushRequest req) {
142 getSystem().getExecutor().execute(new Runnable() {
143 @Override
144 public void run() {
145 try (Repository git = getLeader().openRepository()) {
146 try {
147 push(git, req);
148 req.done(git);
149 } catch (Throwable err) {
150 req.setException(git, err);
151 }
152 } catch (IOException err) {
153 req.setException(null, err);
154 }
155 }
156 });
157 }
158
159 private void push(Repository repo, ReplicaPushRequest req)
160 throws NotSupportedException, TransportException, IOException {
161 Map<String, Ref> adv;
162 List<RemoteCommand> cmds = asUpdateList(req.getCommands());
163 try (Transport transport = Transport.open(repo, uri)) {
164 RemoteConfig rc = getRemoteConfig();
165 if (rc != null) {
166 transport.applyConfig(rc);
167 }
168 transport.setPushAtomic(true);
169 adv = push(repo, transport, cmds);
170 }
171 for (RemoteCommand c : cmds) {
172 c.copyStatusToResult();
173 }
174 req.setRefs(adv);
175 }
176
177 private Map<String, Ref> push(Repository git, Transport transport,
178 List<RemoteCommand> cmds) throws IOException {
179 Map<String, RemoteRefUpdate> updates = asUpdateMap(cmds);
180 try (PushConnection connection = transport.openPush()) {
181 Map<String, Ref> adv = connection.getRefsMap();
182 RemoteRefUpdate accepted = updates.get(getSystem().getTxnAccepted());
183 if (accepted != null && !isExpectedValue(adv, accepted)) {
184 abort(cmds);
185 return adv;
186 }
187
188 RemoteRefUpdate committed = updates.get(getSystem().getTxnCommitted());
189 if (committed != null && !isExpectedValue(adv, committed)) {
190 abort(cmds);
191 return adv;
192 }
193 if (committed != null && getCommitMethod() == ALL_REFS) {
194 prepareCommit(git, cmds, updates, adv,
195 committed.getNewObjectId());
196 }
197
198 connection.push(NullProgressMonitor.INSTANCE, updates);
199 return adv;
200 }
201 }
202
203 private static boolean isExpectedValue(Map<String, Ref> adv,
204 RemoteRefUpdate u) {
205 Ref r = adv.get(u.getRemoteName());
206 if (!AnyObjectId.equals(getId(r), u.getExpectedOldObjectId())) {
207 ((RemoteCommand) u).cmd.setResult(LOCK_FAILURE);
208 return false;
209 }
210 return true;
211 }
212
213 private void prepareCommit(Repository git, List<RemoteCommand> cmds,
214 Map<String, RemoteRefUpdate> updates, Map<String, Ref> adv,
215 ObjectId committed) throws IOException {
216 for (ReceiveCommand cmd : prepareCommit(git, adv, committed)) {
217 RemoteCommand c = new RemoteCommand(cmd);
218 cmds.add(c);
219 updates.put(c.getRemoteName(), c);
220 }
221 }
222
223 private static List<RemoteCommand> asUpdateList(
224 Collection<ReceiveCommand> cmds) {
225 try {
226 List<RemoteCommand> toPush = new ArrayList<>(cmds.size());
227 for (ReceiveCommand cmd : cmds) {
228 toPush.add(new RemoteCommand(cmd));
229 }
230 return toPush;
231 } catch (IOException e) {
232
233 throw new IllegalStateException(e);
234 }
235 }
236
237 private static Map<String, RemoteRefUpdate> asUpdateMap(
238 List<RemoteCommand> cmds) {
239 Map<String, RemoteRefUpdate> m = new LinkedHashMap<>();
240 for (RemoteCommand cmd : cmds) {
241 m.put(cmd.getRemoteName(), cmd);
242 }
243 return m;
244 }
245
246 private static void abort(List<RemoteCommand> cmds) {
247 List<ReceiveCommand> tmp = new ArrayList<>(cmds.size());
248 for (RemoteCommand cmd : cmds) {
249 tmp.add(cmd.cmd);
250 }
251 ReceiveCommand.abort(tmp);
252 }
253
254
255 @Override
256 protected void blockingFetch(Repository repo, ReplicaFetchRequest req)
257 throws NotSupportedException, TransportException {
258 try (Transport transport = Transport.open(repo, uri)) {
259 RemoteConfig rc = getRemoteConfig();
260 if (rc != null) {
261 transport.applyConfig(rc);
262 }
263 fetch(transport, req);
264 }
265 }
266
267 private void fetch(Transport transport, ReplicaFetchRequest req)
268 throws NotSupportedException, TransportException {
269 try (FetchConnection conn = transport.openFetch()) {
270 Map<String, Ref> remoteRefs = conn.getRefsMap();
271 req.setRefs(remoteRefs);
272
273 List<Ref> want = new ArrayList<>();
274 for (String name : req.getWantRefs()) {
275 Ref ref = remoteRefs.get(name);
276 if (ref != null && ref.getObjectId() != null) {
277 want.add(ref);
278 }
279 }
280 for (ObjectId id : req.getWantObjects()) {
281 want.add(new ObjectIdRef.Unpeeled(NETWORK, id.name(), id));
282 }
283
284 conn.fetch(NullProgressMonitor.INSTANCE, want,
285 Collections.<ObjectId> emptySet());
286 }
287 }
288
289 static class RemoteCommand extends RemoteRefUpdate {
290 final ReceiveCommand cmd;
291
292 RemoteCommand(ReceiveCommand cmd) throws IOException {
293 super(null, null,
294 cmd.getNewId(), cmd.getRefName(),
295 true ,
296 null ,
297 cmd.getOldId());
298 this.cmd = cmd;
299 }
300
301 void copyStatusToResult() {
302 if (cmd.getResult() == NOT_ATTEMPTED) {
303 switch (getStatus()) {
304 case OK:
305 case UP_TO_DATE:
306 case NON_EXISTING:
307 cmd.setResult(OK);
308 break;
309
310 case REJECTED_NODELETE:
311 cmd.setResult(REJECTED_NODELETE);
312 break;
313
314 case REJECTED_NONFASTFORWARD:
315 cmd.setResult(REJECTED_NONFASTFORWARD);
316 break;
317
318 case REJECTED_OTHER_REASON:
319 cmd.setResult(REJECTED_OTHER_REASON, getMessage());
320 break;
321
322 default:
323 cmd.setResult(REJECTED_OTHER_REASON, getStatus().name());
324 break;
325 }
326 }
327 }
328 }
329 }