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