1 /***
2 * Copyright 2003-2010 Terracotta, Inc.
3 *
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
7 *
8 * http://www.apache.org/licenses/LICENSE-2.0
9 *
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
15 */
16
17 package net.sf.ehcache.terracotta;
18
19 import static org.mockito.Mockito.mock;
20 import static org.mockito.Mockito.when;
21
22 import java.util.ArrayList;
23 import java.util.Arrays;
24 import java.util.HashMap;
25 import java.util.List;
26 import java.util.Map;
27 import java.util.concurrent.ExecutorService;
28 import java.util.concurrent.Executors;
29
30 import junit.framework.TestCase;
31 import net.sf.ehcache.CacheManager;
32 import net.sf.ehcache.cluster.ClusterNode;
33 import net.sf.ehcache.cluster.ClusterScheme;
34 import net.sf.ehcache.cluster.ClusterTopologyListener;
35
36 import org.junit.Test;
37 import org.junit.runner.RunWith;
38 import org.mockito.runners.MockitoJUnitRunner;
39 import org.slf4j.Logger;
40 import org.slf4j.LoggerFactory;
41
42 @RunWith(MockitoJUnitRunner.class)
43 public class RejoinEventSequenceTest extends TestCase {
44
45 private static final Logger LOG = LoggerFactory.getLogger(RejoinEventSequenceTest.class);
46 private final ExecutorService executor = Executors.newFixedThreadPool(1);
47
48 @Test
49 public void testRejoinEventAfterJoinAndOnline() throws Exception {
50 final ClusteredInstanceFactory mockFactory = mock(ClusteredInstanceFactory.class);
51 final MockCacheCluster mockCacheCluster = new MockCacheCluster();
52
53 TerracottaUnitTesting.setupTerracottaTesting(mockFactory, new Runnable() {
54
55 public void run() {
56 mockCacheCluster.removeAllListeners();
57 }
58
59 });
60 when(mockFactory.getTopology()).thenReturn(mockCacheCluster);
61
62 CacheManager cacheManager = new CacheManager(CacheManager.class.getResourceAsStream("/rejoin/basic-rejoin-test.xml"));
63 RecordingListener listener = new RecordingListener();
64 cacheManager.getCluster(ClusterScheme.TERRACOTTA).addTopologyListener(listener);
65
66
67 int n = 0;
68 while (n < 20) {
69 n++;
70 info("Sleeping for 3 secs...");
71 Thread.sleep(5000);
72 info("================================= Run: " + n + " =========================================================");
73 info("firing node left...");
74
75 mockCacheCluster.fireCurrentNodeLeft();
76
77 info("Sleeping for 3 secs");
78 Thread.sleep(2000);
79 info("Waiting until rejoin");
80 listener.verifyAndWaitUntil(EventType.REJOINED);
81
82 info("Recorded events: ");
83 for (Event e : listener.getEvents()) {
84 info(e.toString());
85 }
86
87 info("Clearing events");
88 listener.clearEvents();
89 }
90 }
91
92 private static void info(String string) {
93 LOG.info("____ " + string);
94 }
95
96 private static final class RecordingListener implements ClusterTopologyListener {
97 private final List<Event> events = new ArrayList<Event>();
98 private volatile EventType state;
99
100 public synchronized void nodeJoined(ClusterNode node) {
101 info("XXX node joined");
102 events.add(new Event(EventType.JOINED));
103 state = EventType.JOINED;
104 notifyAll();
105 }
106
107 public synchronized void clearEvents() {
108 events.clear();
109 }
110
111 public synchronized void nodeLeft(ClusterNode node) {
112 info("XXX node left");
113 events.add(new Event(EventType.LEFT));
114 state = EventType.LEFT;
115 notifyAll();
116 }
117
118 public synchronized void clusterOnline(ClusterNode node) {
119 info("XXX node online");
120 events.add(new Event(EventType.ONLINE));
121 state = EventType.ONLINE;
122 notifyAll();
123 }
124
125 public synchronized void clusterOffline(ClusterNode node) {
126 info("XXX node offline");
127 events.add(new Event(EventType.OFFLINE));
128 state = EventType.OFFLINE;
129 notifyAll();
130 }
131
132 public synchronized void clusterRejoined(ClusterNode oldNode, ClusterNode newNode) {
133 info("XXX node rejoined");
134 events.add(new Event(EventType.REJOINED));
135 state = EventType.REJOINED;
136 notifyAll();
137 }
138
139 public synchronized List<Event> getEvents() {
140 return new ArrayList<RejoinEventSequenceTest.Event>(events);
141 }
142
143 public void verifyAndWaitUntil(EventType type) {
144 synchronized (this) {
145 EventType current = null;
146 for (Event e : events) {
147 verifyTransition(current, e.type);
148 current = e.type;
149 }
150 while (state != type) {
151 try {
152 wait();
153 } catch (InterruptedException e) {
154 e.printStackTrace();
155 }
156 }
157 }
158 }
159
160 private void verifyTransition(EventType current, EventType next) {
161 if (current == null) {
162 return;
163 }
164 if (!EventType.validateTransition(current, next)) {
165 AssertionError error = new AssertionError("Possible transitions from: " + current + " -> "
166 + EventType.getPossibleTransitions(current) + ", but next event received: " + next
167 + ", probably events fired are not in right sequence - " + events);
168 LOG.error("Problem in test", error);
169 throw error;
170 }
171 }
172
173 }
174
175 private static class Event {
176 private final EventType type;
177 private final Thread thread;
178
179 public Event(EventType type) {
180 super();
181 this.type = type;
182 this.thread = Thread.currentThread();
183 }
184
185 @Override
186 public String toString() {
187 return "Event [type=" + type + ", thread=" + thread.getName() + "]";
188 }
189
190 }
191
192 private static enum EventType {
193 JOINED, ONLINE, OFFLINE, LEFT, REJOINED;
194 private static final Map<EventType, List<EventType>> possibleTransitions = new HashMap<RejoinEventSequenceTest.EventType, List<EventType>>();
195 static {
196 possibleTransitions.put(JOINED, Arrays.asList(new EventType[] { EventType.ONLINE }));
197 possibleTransitions.put(ONLINE, Arrays.asList(new EventType[] { EventType.OFFLINE, EventType.REJOINED }));
198 possibleTransitions.put(OFFLINE, Arrays.asList(new EventType[] { EventType.ONLINE, EventType.LEFT }));
199 possibleTransitions.put(LEFT, Arrays.asList(new EventType[] { EventType.JOINED }));
200 possibleTransitions.put(REJOINED, Arrays.asList(new EventType[] { EventType.OFFLINE }));
201 }
202
203 public static boolean validateTransition(EventType from, EventType next) {
204 List<EventType> list = possibleTransitions.get(from);
205 if (list == null) {
206 throw new AssertionError("No possible transitions list for: " + from);
207 }
208 return list.contains(next);
209 }
210
211 public static List<EventType> getPossibleTransitions(EventType from) {
212 return possibleTransitions.get(from);
213 }
214 }
215 }