FilesExpand file tree

 master

/

EventReplayer.java

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

206 lines (184 loc) · 5.25 KB

 master

/

EventReplayer.java

Copy path

Top

File metadata and controls

• Code
• Blame

206 lines (184 loc) · 5.25 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

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

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

import java.util.HashMap;

import java.util.HashSet;

import java.util.Set;

import java.util.concurrent.PriorityBlockingQueue;

import java.util.concurrent.locks.Lock;

import java.util.concurrent.locks.ReentrantLock;

/**

*

* Takes the event recorded by the DocumentEventCapturer and replays

* them in a JTextArea. The delay of 1 sec is only to make the individual

* steps in the reply visible to humans.

*

* @author Jesper Buus Nielsen

*

*/

public class EventReplayer implements Runnable {

private DocumentEventCapturer dec;

private DistributedTextEditor editor;

private PriorityBlockingQueue<TextEvent> eventHistory;

private HashMap<TimeStamp, Set<Integer>> acknowledgements;

private Lock ackLock, eventHistoryLock, caretLock;

private LamportClock lc;

private HashMap<Integer, Integer> carets;

private boolean wasInterrupted = false;

private int maxIdSoFar;

public EventReplayer(DistributedTextEditor editor, DocumentEventCapturer dec, LamportClock lc) {

this.dec = dec;

this.lc = lc;

this.editor = editor;

eventHistory = dec.eventHistory;

acknowledgements = new HashMap<TimeStamp, Set<Integer>>();

carets = new HashMap<Integer, Integer>();

ackLock = new ReentrantLock();

caretLock = new ReentrantLock();

eventHistoryLock = dec.getEventHistoryLock();

}

/**

* When the EventReplayer runs, it empties the eventHistory.

* Whenever an element is taken out of the queue, it checks that the event has been acknowledged by all currently connected peers.

* Then it updates the position of the carets, depending on which event it is.

*/

public void run() {

while (!wasInterrupted) {

if(!eventHistory.isEmpty()) {

TextEvent head = eventHistory.peek();

TimeStamp match = null;

ackLock.lock();

boolean isAck = true;

if(acknowledgements.containsKey(head.getTimeStamp()) || dec.getPeers().isEmpty()) {

Set<Integer> set = acknowledgements.get(head.getTimeStamp());

dec.getPeerLock().lock();

for(Peer p : dec.getPeers()) {

if(!set.contains(p.getId())) {

isAck = false;

break;

}

}

dec.getPeerLock().unlock();

if(isAck && head.equals(eventHistory.peek())) {

eventHistoryLock.lock();

caretLock.lock();

try {

int pos = 0;

TextEvent e = eventHistory.take();

match = head.getTimeStamp();

if(head instanceof CaretUpdate) {

pos = ((CaretUpdate) head).getPos();

} else {

int idOfSender = e.getTimeStamp().getID();

pos = carets.get(idOfSender);

}

e.doEvent(editor, pos);

updateAllCarets(e, pos);

} catch (InterruptedException e1) {

e1.printStackTrace();

}

caretLock.unlock();

eventHistoryLock.unlock();

}

}

if(match != null) {

acknowledgements.remove(match);

}

ackLock.unlock();

}

}

System.out.println("I'm the thread running the EventReplayer, now I die!");

}

/**

* Will send null to the other peers if the connection is not closed.

*/

public void stopStreamToQueue() {

dec.getPeerLock().lock();

for(Peer p : dec.getPeers()) {

if (p.isConnected()) {

p.writeObjectToStream(null);

}

}

dec.getPeerLock().unlock();

wasInterrupted = true;

}

public void updateCaretPos(int id, int pos) {

carets.put(id, pos);

}

private void updateAllCarets(TextEvent e, int pos) {

if(e instanceof TextInsertEvent) {

for(int i : carets.keySet()) {

if(carets.get(i) >= pos) {

carets.put(i, carets.get(i) + e.getLength());

}

}

} else if (e instanceof TextRemoveEvent) {

for(int i : carets.keySet()) {

if(carets.get(i) >= pos) {

carets.put(i, carets.get(i) - e.getLength());

}

}

}

}

public PriorityBlockingQueue<TextEvent> getEventHistory() {

return eventHistory;

}

public Lock getMapLock() {

return ackLock;

}

public Lock getEventHistoryLock() {

return eventHistoryLock;

}

public LamportClock getLc() {

return lc;

}

public DocumentEventCapturer getDocumentEventCapturer() {

return dec;

}

public void addAcknowledgement(TimeStamp ts, int id) {

ackLock.lock();

if(!acknowledgements.containsKey(ts)) {

acknowledgements.put(ts, new HashSet<Integer>());

}

acknowledgements.get(ts).add(id);

ackLock.unlock();

}

public HashMap<TimeStamp, Set<Integer>> getAcknowledgements() {

return acknowledgements;

}

public HashMap<Integer, Integer> getCarets() {

return carets;

}

public void setEventHistory(PriorityBlockingQueue<TextEvent> eventHistory) {

eventHistoryLock.lock();

for(TextEvent e : eventHistory) {

this.eventHistory.put(e);

}

eventHistoryLock.unlock();

}

public void setAcknowledgements(

HashMap<TimeStamp, Set<Integer>> acknowledgements) {

ackLock.lock();

this.acknowledgements = acknowledgements;

ackLock.unlock();

}

public void setCarets(HashMap<Integer, Integer> carets) {

caretLock.lock();

this.carets = carets;

caretLock.unlock();

}

public void addCaretPos(int id, int caretPos) {

caretLock.lock();

carets.put(id, caretPos);

caretLock.unlock();

}

public void removePeer(Peer peer) {

dec.getPeerLock().lock();

dec.getPeers().remove(peer);

dec.updateConnectionStatusArea();

dec.getPeerLock().unlock();

caretLock.lock();

carets.remove(peer.getId());

caretLock.unlock();

}

public Lock getCaretLock() {

return caretLock;

}

}