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
177
178
179
|
/****************************************************************************
* Copyright (c) 2008 Mustafa K. Isik and others.
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* which accompanies this distribution, and is available at
* http://www.eclipse.org/legal/epl-v10.html
*
* Contributors:
* Mustafa K. Isik - initial API and implementation
*****************************************************************************/
package org.eclipse.ecf.internal.provisional.docshare.cola;
import java.util.*;
import org.eclipse.core.runtime.Assert;
import org.eclipse.ecf.core.util.Trace;
import org.eclipse.ecf.internal.docshare.Activator;
import org.eclipse.ecf.internal.docshare.DocshareDebugOptions;
import org.eclipse.ecf.internal.provisional.docshare.DocShare;
import org.eclipse.ecf.internal.provisional.docshare.SynchronizationStrategy;
import org.eclipse.ecf.internal.provisional.docshare.messages.UpdateMessage;
import org.eclipse.osgi.util.NLS;
public class ColaSynchronizer implements SynchronizationStrategy {
// <ColaUpdateMessage>
private final LinkedList unacknowledgedLocalOperations;
private final boolean isInitiator;
private long localOperationsCount;
private long remoteOperationsCount;
// <DocShare, ColaSynchronizer>
private static Map sessionStrategies = new HashMap();
private ColaSynchronizer(DocShare docshare) {
this.isInitiator = docshare.isInitiator();
unacknowledgedLocalOperations = new LinkedList();
localOperationsCount = 0;
remoteOperationsCount = 0;
}
public static ColaSynchronizer getInstanceFor(DocShare docshare) {
if (sessionStrategies.get(docshare) == null) {
sessionStrategies.put(docshare, new ColaSynchronizer(docshare));
}
return (ColaSynchronizer) sessionStrategies.get(docshare);
}
public static void cleanUpFor(DocShare docshare) {
sessionStrategies.remove(docshare);
}
public UpdateMessage registerOutgoingMessage(UpdateMessage localMsg) {
Trace.entering(Activator.PLUGIN_ID, DocshareDebugOptions.METHODS_ENTERING, this.getClass(), "registerOutgoingMessage", localMsg); //$NON-NLS-1$
final ColaUpdateMessage colaMsg = new ColaUpdateMessage(localMsg, localOperationsCount, remoteOperationsCount);
if (!colaMsg.isReplacement()) {
unacknowledgedLocalOperations.add(colaMsg);
localOperationsCount++;
}
Trace.exiting(Activator.PLUGIN_ID, DocshareDebugOptions.METHODS_EXITING, this.getClass(), "registerOutgoingMessage", colaMsg); //$NON-NLS-1$
return colaMsg;
}
/**
* Handles proper transformation of incoming <code>ColaUpdateMessage</code>s.
* Returned <code>UpdateMessage</code>s can be applied directly to the
* shared document. The method implements the concurrency algorithm described
* in <code>http://wiki.eclipse.org/RT_Shared_Editing</code>
* @param remoteMsg
* @return List contains <code>UpdateMessage</code>s ready for sequential application to document
*/
public List transformIncomingMessage(final UpdateMessage remoteMsg) {
if (!(remoteMsg instanceof ColaUpdateMessage)) {
throw new IllegalArgumentException("UpdateMessage is incompatible with Cola SynchronizationStrategy"); //$NON-NLS-1$
}
Trace.entering(Activator.PLUGIN_ID, DocshareDebugOptions.METHODS_ENTERING, this.getClass(), "transformIncomingMessage", remoteMsg); //$NON-NLS-1$
ColaUpdateMessage transformedRemote = (ColaUpdateMessage) remoteMsg;
List transformedRemotes = new LinkedList();
transformedRemotes.add(transformedRemote);
remoteOperationsCount++;
//this is where the concurrency algorithm is executed
if (!unacknowledgedLocalOperations.isEmpty()) {//Do not remove this. It is necessary. The following iterator does not suffice.
// remove operations from queue that have been implicitly
// acknowledged as received on the remote site by the reception of
// this message
for (final Iterator it = unacknowledgedLocalOperations.iterator(); it.hasNext();) {
ColaUpdateMessage unackedLocalOp = (ColaUpdateMessage) it.next();
if (transformedRemote.getRemoteOperationsCount() > unackedLocalOp.getLocalOperationsCount()) {
Trace.trace(Activator.PLUGIN_ID, NLS.bind("transformIncomingMessage.removing {0}", unackedLocalOp)); //$NON-NLS-1$
it.remove();
} else {
// the unackowledgedLocalOperations queue is ordered and
// sorted
// due to sequential insertion of local ops, thus once a
// local op with a higher
// or equal local op count (i.e. remote op count from the
// remote operation's view)
// is reached, we can abandon the check for the remaining
// queue items
Trace.trace(Activator.PLUGIN_ID, "breaking out of unackedLocalOperations loop"); //$NON-NLS-1$
break;// exits for-loop
}
}
// at this point the queue has been freed of operations that
// don't require to be transformed against
if (!unacknowledgedLocalOperations.isEmpty()) {
ColaUpdateMessage localOp = (ColaUpdateMessage) unacknowledgedLocalOperations.getFirst();
Assert.isTrue(transformedRemote.getRemoteOperationsCount() == localOp.getLocalOperationsCount());
for (final ListIterator unackOpsListIt = unacknowledgedLocalOperations.listIterator(); unackOpsListIt.hasNext();) {
for (final ListIterator trafoRemotesIt = transformedRemotes.listIterator(); trafoRemotesIt.hasNext();) {
// returns new instance
// clarify operation preference, owner/docshare initiator
// consistently comes first
localOp = (ColaUpdateMessage) unackOpsListIt.next();
transformedRemote = (ColaUpdateMessage) trafoRemotesIt.next();
transformedRemote = transformedRemote.transformAgainst(localOp, isInitiator);
if (transformedRemote.isSplitUp()) {
//currently this only happens for a remote deletion that needs to be transformed against a locally applied insertion
//attention: before applying a list of deletions to docshare, the indices need to be updated/finalized one last time
//since deletions can only be applied sequentially and every deletion is going to change the underlying document and the
//respective indices!
trafoRemotesIt.remove();
for (final Iterator splitUpIterator = transformedRemote.getSplitUpRepresentation().iterator(); splitUpIterator.hasNext();) {
trafoRemotesIt.add(splitUpIterator.next());
}
//according to the ListIterator documentation it seems so as if the following line is unnecessary,
//as a call to next() after the last removal and additions will return what it would have returned anyway
//trafoRemotesIt.next();//TODO not sure about the need for this - I want to jump over the two inserted ops and reach the end of this iterator
}
//TODO check whether or not this collection shuffling does what it is supposed to, i.e. remove current localop in unack list and add split up representation instead
if (localOp.isSplitUp()) {
//local operation has been split up during operational transform --> remove current version and add new versions plus jump over those
unackOpsListIt.remove();
for (final Iterator splitUpOpIterator = localOp.getSplitUpRepresentation().iterator(); splitUpOpIterator.hasNext();) {
unackOpsListIt.add(splitUpOpIterator.next());
}
//according to the ListIterator documentation it seems so as if the following line is unnecessary,
//as a call to next() after the last removal and additions will return what it would have returned anyway
//unackOpsListIt.next();//TODO check whether or not this does jump over both inserted operations that replaced the current unack op
}//end split up localop handling
}//transformedRemotes List iteration
}
}
}
Trace.exiting(Activator.PLUGIN_ID, DocshareDebugOptions.METHODS_EXITING, this.getClass(), "transformIncomingMessage", transformedRemote); //$NON-NLS-1$
//TODO find a cleaner and more OO way of cleaning up the list if it contains multiple deletions:
if (transformedRemotes.size() > 1) {
ColaUpdateMessage firstOp = (ColaUpdateMessage) transformedRemotes.get(0);
if (firstOp.isDeletion()) {
//this means all operations in the return list must also be deletions, i.e. modify virtual/optimistic offset for sequential application to document
ListIterator deletionFinalizerIt = transformedRemotes.listIterator();
ColaUpdateMessage previousDel = (ColaUpdateMessage) deletionFinalizerIt.next();//jump over first del-op does not need modification, we know this is OK because of previous size check;
ColaUpdateMessage currentDel;
for (; deletionFinalizerIt.hasNext();) {
currentDel = (ColaUpdateMessage) deletionFinalizerIt.next();
currentDel.setOffset(currentDel.getOffset() - previousDel.getLengthOfReplacedText());
previousDel = currentDel;
}
}
}
return transformedRemotes;
}
public String toString() {
StringBuffer buf = new StringBuffer("ColaSynchronizer"); //$NON-NLS-1$
return buf.toString();
}
}
|
