mirror of
https://github.com/eclipse-sailing-analytics/sailing-analytics.git
synced 2026-11-02 17:19:37 +00:00
introduced replication test support bundle, factoring commonalities for abstract replication tests; started simple security replication test frame
This commit is contained in:
1 parent
801946302f
commit
6fc64beb42
26 files changed
+535
-291
No files matched your search
@@ -23,4 +23,5 @@ Require-Bundle: com.sap.sailing.server,
|
||||
org.hamcrest;bundle-version="1.1.0",
|
||||
com.sap.sse.common,
|
||||
org.mockito.mockito-core;bundle-version="1.9.5",
|
||||
com.sap.sailing.domain.shared.android
|
||||
com.sap.sailing.domain.shared.android,
|
||||
com.sap.sse.replication.testsupport
|
||||
+17
-261
@@ -1,32 +1,5 @@
|
||||
package com.sap.sailing.server.replication.test;
|
||||
|
||||
import static org.junit.Assert.assertFalse;
|
||||
|
||||
import java.io.BufferedReader;
|
||||
import java.io.IOException;
|
||||
import java.io.InputStreamReader;
|
||||
import java.io.ObjectInputStream;
|
||||
import java.io.ObjectOutputStream;
|
||||
import java.io.OutputStreamWriter;
|
||||
import java.io.PipedInputStream;
|
||||
import java.io.PipedOutputStream;
|
||||
import java.io.PrintWriter;
|
||||
import java.net.ConnectException;
|
||||
import java.net.InetAddress;
|
||||
import java.net.ServerSocket;
|
||||
import java.net.Socket;
|
||||
import java.net.URL;
|
||||
import java.net.URLConnection;
|
||||
import java.util.UUID;
|
||||
import java.util.zip.GZIPOutputStream;
|
||||
|
||||
import org.junit.After;
|
||||
import org.junit.Before;
|
||||
import org.junit.Rule;
|
||||
import org.junit.rules.Timeout;
|
||||
|
||||
import com.rabbitmq.client.Channel;
|
||||
import com.rabbitmq.client.QueueingConsumer;
|
||||
import com.sap.sailing.domain.base.DomainFactory;
|
||||
import com.sap.sailing.domain.base.impl.DomainFactoryImpl;
|
||||
import com.sap.sailing.domain.persistence.MongoObjectFactory;
|
||||
@@ -36,256 +9,39 @@ import com.sap.sailing.domain.racelog.tracking.EmptyGPSFixStore;
|
||||
import com.sap.sailing.domain.tracking.impl.EmptyWindStore;
|
||||
import com.sap.sailing.server.RacingEventService;
|
||||
import com.sap.sailing.server.impl.RacingEventServiceImpl;
|
||||
import com.sap.sse.common.Util;
|
||||
import com.sap.sse.mongodb.MongoDBService;
|
||||
import com.sap.sse.replication.Replicable;
|
||||
import com.sap.sse.replication.ReplicationMasterDescriptor;
|
||||
import com.sap.sse.replication.ReplicationService;
|
||||
import com.sap.sse.replication.impl.Activator;
|
||||
import com.sap.sse.replication.impl.RabbitOutputStream;
|
||||
import com.sap.sse.replication.impl.ReplicaDescriptor;
|
||||
import com.sap.sse.replication.impl.ReplicationInstancesManager;
|
||||
import com.sap.sse.replication.impl.ReplicationMasterDescriptorImpl;
|
||||
import com.sap.sse.replication.impl.ReplicationServiceImpl;
|
||||
import com.sap.sse.replication.impl.Replicator;
|
||||
import com.sap.sse.replication.impl.SingletonReplicablesProvider;
|
||||
|
||||
public abstract class AbstractServerReplicationTest {
|
||||
protected static final int SERVLET_PORT = 9990;
|
||||
protected RacingEventServiceImpl replica;
|
||||
protected RacingEventServiceImpl master;
|
||||
protected ReplicationServiceTestImpl replicaReplicator;
|
||||
private ReplicaDescriptor replicaDescriptor;
|
||||
private ReplicationServiceImpl masterReplicator;
|
||||
private ReplicationMasterDescriptor masterDescriptor;
|
||||
public abstract class AbstractServerReplicationTest extends com.sap.sse.replication.testsupport.AbstractServerReplicationTest<RacingEventService, RacingEventServiceImpl> {
|
||||
protected MongoDBService mongoDBService;
|
||||
protected MongoObjectFactory mongoObjectFactory;
|
||||
|
||||
@Rule public Timeout AbstractTracTracLiveTestTimeout = new Timeout(5 * 60 * 1000); // timeout after 5 minutes
|
||||
private Thread initialLoadTestServerThread;
|
||||
|
||||
/**
|
||||
* Drops the test DB. Sets up master and replica, starts the JMS message broker and registers the replica with the master.
|
||||
* Drops the test DB, if <code>dropDB</code> is <code>true</code> and requests the DB to start.
|
||||
*/
|
||||
@Before
|
||||
public void setUp() throws Exception {
|
||||
try {
|
||||
Util.Pair<ReplicationServiceTestImpl, ReplicationMasterDescriptor> result = basicSetUp(
|
||||
true, /* master=null means create a new one */ null,
|
||||
/* replica=null means create a new one */null);
|
||||
result.getA().startToReplicateFrom(result.getB());
|
||||
} catch (Exception e) {
|
||||
tearDown();
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Drops the test DB.
|
||||
*
|
||||
* @param master
|
||||
* if not <code>null</code>, the value will be used for {@link #master}; otherwise, a new racing event
|
||||
* service will be created as master
|
||||
* @param replica
|
||||
* if not <code>null</code>, the value will be used for {@link #replica}; otherwise, a new racing event
|
||||
* service will be created as replica
|
||||
*/
|
||||
protected Util.Pair<ReplicationServiceTestImpl, ReplicationMasterDescriptor> basicSetUp(
|
||||
boolean dropDB, RacingEventServiceImpl master, RacingEventServiceImpl replica) throws IOException, InterruptedException {
|
||||
String exchangeName = "test-sapsailinganalytics-exchange";
|
||||
String exchangeHost = "localhost";
|
||||
if (System.getenv(Activator.ENV_VAR_NAME_REPLICATION_HOST) != null) {
|
||||
exchangeHost = System.getenv(Activator.ENV_VAR_NAME_REPLICATION_HOST);
|
||||
}
|
||||
final UUID serverUuid = UUID.randomUUID();
|
||||
final MongoDBService mongoDBService = MongoDBService.INSTANCE;
|
||||
@Override
|
||||
protected void persistenceSetUp(boolean dropDB) {
|
||||
mongoDBService = MongoDBService.INSTANCE;
|
||||
if (dropDB) {
|
||||
mongoDBService.getDB().dropDatabase();
|
||||
}
|
||||
final MongoObjectFactory mongoObjectFactory = PersistenceFactory.INSTANCE.getMongoObjectFactory(mongoDBService);
|
||||
mongoObjectFactory = PersistenceFactory.INSTANCE.getMongoObjectFactory(mongoDBService);
|
||||
mongoObjectFactory.getDatabase().requestStart();
|
||||
if (master != null) {
|
||||
this.master = master;
|
||||
} else {
|
||||
this.master = createNewMaster(mongoDBService, mongoObjectFactory);
|
||||
}
|
||||
if (replica != null) {
|
||||
this.replica = replica;
|
||||
} else {
|
||||
this.replica = new RacingEventServiceImpl(PersistenceFactory.INSTANCE.getDomainObjectFactory(mongoDBService,
|
||||
// replica gets its own base DomainFactory:
|
||||
new DomainFactoryImpl()), mongoObjectFactory, MediaDBFactory.INSTANCE.getMediaDB(mongoDBService), EmptyWindStore.INSTANCE, EmptyGPSFixStore.INSTANCE);
|
||||
}
|
||||
ReplicationInstancesManager rim = new ReplicationInstancesManager();
|
||||
masterReplicator = new ReplicationServiceImpl(exchangeName, exchangeHost, 0, rim, new SingletonReplicablesProvider(this.master));
|
||||
replicaDescriptor = new ReplicaDescriptor(InetAddress.getLocalHost(), serverUuid, "");
|
||||
masterReplicator.registerReplica(replicaDescriptor);
|
||||
// connect to exchange host and local server running as master
|
||||
// master server and exchange host can be two different hosts
|
||||
masterDescriptor = new ReplicationMasterDescriptorImpl(exchangeHost, exchangeName, 0, UUID.randomUUID().toString(), "localhost", SERVLET_PORT);
|
||||
ReplicationServiceTestImpl replicaReplicator = new ReplicationServiceTestImpl(exchangeName, exchangeHost, rim, replicaDescriptor,
|
||||
this.replica, this.master, masterReplicator, masterDescriptor);
|
||||
Util.Pair<ReplicationServiceTestImpl, ReplicationMasterDescriptor> result = new Util.Pair<>(replicaReplicator, masterDescriptor);
|
||||
initialLoadTestServerThread = replicaReplicator.startInitialLoadTransmissionServlet();
|
||||
this.replicaReplicator = replicaReplicator;
|
||||
return result;
|
||||
}
|
||||
|
||||
protected RacingEventServiceImpl createNewMaster(final MongoDBService mongoDBService,
|
||||
final MongoObjectFactory mongoObjectFactory) {
|
||||
protected RacingEventServiceImpl createNewMaster() {
|
||||
return new RacingEventServiceImpl(PersistenceFactory.INSTANCE.getDomainObjectFactory(mongoDBService,
|
||||
DomainFactory.INSTANCE), mongoObjectFactory, MediaDBFactory.INSTANCE.getMediaDB(mongoDBService),
|
||||
EmptyWindStore.INSTANCE, EmptyGPSFixStore.INSTANCE);
|
||||
}
|
||||
|
||||
protected RacingEventServiceImpl createNewReplica() {
|
||||
return new RacingEventServiceImpl(PersistenceFactory.INSTANCE.getDomainObjectFactory(mongoDBService,
|
||||
// replica gets its own base DomainFactory:
|
||||
new DomainFactoryImpl()), mongoObjectFactory, MediaDBFactory.INSTANCE.getMediaDB(mongoDBService),
|
||||
EmptyWindStore.INSTANCE, EmptyGPSFixStore.INSTANCE);
|
||||
}
|
||||
|
||||
@After
|
||||
public void tearDown() throws Exception {
|
||||
final MongoDBService mongoDBService = MongoDBService.INSTANCE;
|
||||
final MongoObjectFactory mongoObjectFactory = PersistenceFactory.INSTANCE.getMongoObjectFactory(mongoDBService);
|
||||
protected void persistenceTearDown() {
|
||||
mongoObjectFactory.getDatabase().requestDone();
|
||||
if (masterReplicator != null) {
|
||||
masterReplicator.unregisterReplica(replicaDescriptor);
|
||||
}
|
||||
if (masterDescriptor != null) {
|
||||
masterDescriptor.stopConnection();
|
||||
}
|
||||
try {
|
||||
if (initialLoadTestServerThread != null) {
|
||||
URLConnection urlConnection = new URL("http://localhost:"+SERVLET_PORT+"/STOP").openConnection(); // stop the initial load test server thread
|
||||
urlConnection.getInputStream().close();
|
||||
initialLoadTestServerThread.join(10000 /* wait 10s */);
|
||||
assertFalse("Expected initial load test server thread to die", initialLoadTestServerThread.isAlive());
|
||||
}
|
||||
} catch (ConnectException ex) {
|
||||
/* do not make tests fail because of a server that has been shut down
|
||||
* or when an exception occured (see setUp()) - let the
|
||||
* original exception propagate */
|
||||
ex.printStackTrace();
|
||||
}
|
||||
}
|
||||
|
||||
public void stopReplicatingToMaster() throws IOException {
|
||||
replicaReplicator.stopToReplicateFromMaster();
|
||||
}
|
||||
|
||||
static class ReplicationServiceTestImpl extends ReplicationServiceImpl {
|
||||
protected static final int INITIAL_LOAD_PACKAGE_SIZE = 1024*1024;
|
||||
private final RacingEventService master;
|
||||
private final ReplicaDescriptor replicaDescriptor;
|
||||
private final ReplicationService masterReplicationService;
|
||||
private final ReplicationMasterDescriptor masterDescriptor;
|
||||
|
||||
public ReplicationServiceTestImpl(String exchangeName, String exchangeHost, ReplicationInstancesManager replicationInstancesManager,
|
||||
ReplicaDescriptor replicaDescriptor, RacingEventService replica,
|
||||
RacingEventService master, ReplicationService masterReplicationService, ReplicationMasterDescriptor masterDescriptor)
|
||||
throws IOException {
|
||||
super(exchangeName, exchangeHost, 0, replicationInstancesManager, new SingletonReplicablesProvider(replica));
|
||||
this.replicaDescriptor = replicaDescriptor;
|
||||
this.master = master;
|
||||
this.masterReplicationService = masterReplicationService;
|
||||
this.masterDescriptor = masterDescriptor;
|
||||
}
|
||||
|
||||
private Thread startInitialLoadTransmissionServlet() throws InterruptedException {
|
||||
final boolean[] listening = new boolean[] { false };
|
||||
Thread initialLoadTestServerThread = new Thread("Replication initial load test server") {
|
||||
public void run() {
|
||||
ServerSocket ss;
|
||||
try {
|
||||
ss = new ServerSocket(SERVLET_PORT);
|
||||
synchronized (listening) {
|
||||
listening[0] = true;
|
||||
listening.notifyAll();
|
||||
}
|
||||
boolean stop = false;
|
||||
while (!stop) {
|
||||
Socket s = ss.accept();
|
||||
String request = new BufferedReader(new InputStreamReader(s.getInputStream())).readLine();
|
||||
PrintWriter pw = new PrintWriter(new OutputStreamWriter(s.getOutputStream()));
|
||||
pw.println("HTTP/1.1 200 OK");
|
||||
pw.println("Content-Type: text/plain");
|
||||
pw.println();
|
||||
pw.flush();
|
||||
if (request.contains("DEREGISTER")) {
|
||||
// assuming that it is safe to unregister all replicas for tests
|
||||
for (ReplicaDescriptor descriptor : getReplicaInfo()) {
|
||||
unregisterReplica(descriptor);
|
||||
}
|
||||
} else if (request.contains("REGISTER")) {
|
||||
final String uuid = UUID.randomUUID().toString();
|
||||
registerReplicaUuidForMaster(uuid, masterDescriptor);
|
||||
pw.print(uuid.getBytes());
|
||||
} else if (request.contains("INITIAL_LOAD")) {
|
||||
Channel channel = masterReplicationService.createMasterChannel();
|
||||
RabbitOutputStream ros = new RabbitOutputStream(INITIAL_LOAD_PACKAGE_SIZE, channel,
|
||||
/* queueName */ "initial-load-for-TestClient-"+UUID.randomUUID(), /* syncAfterTimeout */ false);
|
||||
pw.println(ros.getQueueName());
|
||||
final GZIPOutputStream gzipOutputStream = new GZIPOutputStream(ros);
|
||||
master.serializeForInitialReplication(gzipOutputStream);
|
||||
gzipOutputStream.finish();
|
||||
ros.close();
|
||||
} else if (request.contains("STOP")) {
|
||||
stop = true;
|
||||
}
|
||||
pw.close();
|
||||
s.close();
|
||||
}
|
||||
ss.close();
|
||||
} catch (IOException e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
};
|
||||
initialLoadTestServerThread.start();
|
||||
synchronized (listening) {
|
||||
while (!listening[0]) {
|
||||
listening.wait();
|
||||
}
|
||||
}
|
||||
return initialLoadTestServerThread;
|
||||
}
|
||||
|
||||
protected Replicator startToReplicateFromButDontYetFetchInitialLoad(ReplicationMasterDescriptor master, boolean startReplicatorSuspended)
|
||||
throws IOException {
|
||||
masterReplicationService.registerReplica(replicaDescriptor);
|
||||
registerReplicaUuidForMaster(replicaDescriptor.getUuid().toString(), master);
|
||||
QueueingConsumer consumer = master.getConsumer();
|
||||
final Replicator replicator = new Replicator(master, getReplicablesProvider(), startReplicatorSuspended, consumer);
|
||||
new Thread(replicator).start();
|
||||
return replicator;
|
||||
}
|
||||
|
||||
/**
|
||||
* Clones the {@link #master}'s state to the {@link #replica} using
|
||||
* {@link RacingEventServiceImpl#serializeForInitialReplication(ObjectOutputStream)} and
|
||||
* {@link RacingEventServiceImpl#initiallyFillFrom(ObjectInputStream)} through a piped input/output stream.
|
||||
* @throws InterruptedException
|
||||
*/
|
||||
protected void initialLoad() throws IOException, ClassNotFoundException, InterruptedException {
|
||||
PipedOutputStream pos = new PipedOutputStream();
|
||||
PipedInputStream pis = new PipedInputStream(pos);
|
||||
new Thread("clone writer") {
|
||||
public void run() {
|
||||
try {
|
||||
master.serializeForInitialReplication(pos);
|
||||
pos.close();
|
||||
} catch (IOException e) {
|
||||
e.printStackTrace();
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
}.start();
|
||||
for (Replicable<?, ?> replicable : getReplicablesProvider().getReplicables()) {
|
||||
replicable.clearReplicaState();
|
||||
replicable.initiallyFillFrom(pis);
|
||||
}
|
||||
pis.close();
|
||||
}
|
||||
|
||||
public void waitUntilQueueIsEmpty() throws InterruptedException, IllegalAccessException {
|
||||
synchronized (getReplicator()) {
|
||||
while (!getReplicator().isQueueEmpty()) {
|
||||
getReplicator().wait();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
+7
-9
@@ -22,8 +22,10 @@ import com.rabbitmq.client.ConsumerCancelledException;
|
||||
import com.rabbitmq.client.QueueingConsumer;
|
||||
import com.rabbitmq.client.ShutdownSignalException;
|
||||
import com.sap.sailing.domain.base.Event;
|
||||
import com.sap.sailing.server.RacingEventService;
|
||||
import com.sap.sse.common.TimePoint;
|
||||
import com.sap.sse.common.Util;
|
||||
import com.sap.sse.common.Util.Pair;
|
||||
import com.sap.sse.common.impl.MillisecondsTimePoint;
|
||||
import com.sap.sse.replication.ReplicationMasterDescriptor;
|
||||
import com.sap.sse.replication.impl.ReplicationMasterDescriptorImpl;
|
||||
@@ -79,17 +81,13 @@ public class ConnectionResetAndReconnectTest extends AbstractServerReplicationTe
|
||||
|
||||
}
|
||||
|
||||
private MasterReplicationDescriptorMock masterReplicationDescriptor;
|
||||
private ReplicationServiceTestImpl replicaReplicationDescriptor;
|
||||
|
||||
@Before
|
||||
@Override
|
||||
public void setUp() throws Exception {
|
||||
try {
|
||||
com.sap.sse.common.Util.Pair<ReplicationServiceTestImpl, ReplicationMasterDescriptor> result = basicSetUp(
|
||||
true, /* master=null means create a new one */ null,
|
||||
/* replica=null means create a new one */null);
|
||||
masterReplicationDescriptor = MasterReplicationDescriptorMock.from(result.getB());
|
||||
replicaReplicationDescriptor = result.getA();
|
||||
Pair<com.sap.sse.replication.testsupport.AbstractServerReplicationTest.ReplicationServiceTestImpl<RacingEventService>, ReplicationMasterDescriptor> result = basicSetUp(/* dropDB */ true,
|
||||
/* master=null means create a new one */ null, /* replica=null means create a new one */null);
|
||||
masterDescriptor = MasterReplicationDescriptorMock.from(result.getB());
|
||||
} catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
tearDown();
|
||||
@@ -104,7 +102,7 @@ public class ConnectionResetAndReconnectTest extends AbstractServerReplicationTe
|
||||
/* until here both instances should have the same in-memory state.
|
||||
* now lets add an event on master and stop the messaging queue. */
|
||||
stopMessagingExchange();
|
||||
replicaReplicationDescriptor.startToReplicateFrom(masterReplicationDescriptor);
|
||||
replicaReplicator.startToReplicateFrom(masterDescriptor);
|
||||
Event event = addEventOnMaster();
|
||||
Thread.sleep(1000); // wait for master queue to get filled
|
||||
assertNull(replica.getEvent(event.getId()));
|
||||
|
||||
+4
-1
@@ -34,12 +34,14 @@ import com.sap.sailing.domain.test.mock.MockedTrackedRaceWithFixedRank;
|
||||
import com.sap.sailing.domain.tracking.DynamicTrackedRace;
|
||||
import com.sap.sailing.domain.tracking.TrackedRace;
|
||||
import com.sap.sailing.domain.tracking.impl.EmptyWindStore;
|
||||
import com.sap.sailing.server.RacingEventService;
|
||||
import com.sap.sailing.server.impl.RacingEventServiceImpl;
|
||||
import com.sap.sailing.server.operationaltransformation.AddColumnToLeaderboard;
|
||||
import com.sap.sailing.server.operationaltransformation.ConnectTrackedRaceToLeaderboardColumn;
|
||||
import com.sap.sailing.server.operationaltransformation.CreateFlexibleLeaderboard;
|
||||
import com.sap.sailing.server.operationaltransformation.UpdateLeaderboardMaxPointsReason;
|
||||
import com.sap.sse.common.Util;
|
||||
import com.sap.sse.common.Util.Pair;
|
||||
import com.sap.sse.common.impl.MillisecondsTimePoint;
|
||||
import com.sap.sse.mongodb.MongoDBService;
|
||||
import com.sap.sse.replication.ReplicationMasterDescriptor;
|
||||
@@ -113,7 +115,8 @@ public class DelayedLeaderboardCorrectionsReplicationTest extends AbstractServer
|
||||
masterLeaderboardReloaded.getRaceColumnByName(Q2), MillisecondsTimePoint.now()));
|
||||
|
||||
// replicate the re-loaded environment
|
||||
com.sap.sse.common.Util.Pair<ReplicationServiceTestImpl, ReplicationMasterDescriptor> descriptors = basicSetUp(/* dropDB */ false, master, replica);
|
||||
Pair<com.sap.sse.replication.testsupport.AbstractServerReplicationTest.ReplicationServiceTestImpl<RacingEventService>, ReplicationMasterDescriptor> descriptors = basicSetUp(
|
||||
/* dropDB */false, master, replica);
|
||||
replicaReplicator = descriptors.getA();
|
||||
masterDescriptor = descriptors.getB();
|
||||
// starting to replicate will clear the competitor store used by the replica in this test;
|
||||
|
||||
+7
-8
@@ -42,37 +42,37 @@ import com.sap.sailing.domain.persistence.media.MediaDBFactory;
|
||||
import com.sap.sailing.domain.racelog.tracking.EmptyGPSFixStore;
|
||||
import com.sap.sailing.domain.test.TrackBasedTest;
|
||||
import com.sap.sailing.domain.tracking.impl.EmptyWindStore;
|
||||
import com.sap.sailing.server.RacingEventService;
|
||||
import com.sap.sailing.server.impl.RacingEventServiceImpl;
|
||||
import com.sap.sailing.server.operationaltransformation.AddDefaultRegatta;
|
||||
import com.sap.sailing.server.operationaltransformation.AddRaceDefinition;
|
||||
import com.sap.sailing.server.operationaltransformation.CreateFlexibleLeaderboard;
|
||||
import com.sap.sse.common.TimePoint;
|
||||
import com.sap.sse.common.Util;
|
||||
import com.sap.sse.common.Util.Pair;
|
||||
import com.sap.sse.common.impl.MillisecondsDurationImpl;
|
||||
import com.sap.sse.common.impl.MillisecondsTimePoint;
|
||||
import com.sap.sse.mongodb.MongoDBService;
|
||||
import com.sap.sse.replication.ReplicationMasterDescriptor;
|
||||
import com.sap.sse.replication.impl.Replicator;
|
||||
|
||||
public class InitialLoadReplicationObjectIdentityTest extends AbstractServerReplicationTest {
|
||||
private Util.Pair<ReplicationServiceTestImpl, ReplicationMasterDescriptor> replicationDescriptorPair;
|
||||
private Pair<com.sap.sse.replication.testsupport.AbstractServerReplicationTest.ReplicationServiceTestImpl<RacingEventService>, ReplicationMasterDescriptor> replicationDescriptorPair;
|
||||
|
||||
/**
|
||||
* Drops the test DB. Sets up master and replica, starts the JMS message broker and registers the replica with the master.
|
||||
*/
|
||||
@Before
|
||||
@Override
|
||||
public void setUp() throws Exception {
|
||||
final MongoDBService mongoDBService = MongoDBService.INSTANCE;
|
||||
mongoDBService.getDB().dropDatabase();
|
||||
persistenceSetUp(/* dropDB */ true);
|
||||
this.master = new RacingEventServiceImpl(PersistenceFactory.INSTANCE.getDomainObjectFactory(mongoDBService, DomainFactory.INSTANCE), PersistenceFactory.INSTANCE
|
||||
.getMongoObjectFactory(mongoDBService), MediaDBFactory.INSTANCE.getMediaDB(mongoDBService), EmptyWindStore.INSTANCE, EmptyGPSFixStore.INSTANCE);
|
||||
this.replica = new RacingEventServiceImpl(PersistenceFactory.INSTANCE.getDomainObjectFactory(mongoDBService, DomainFactory.INSTANCE), PersistenceFactory.INSTANCE
|
||||
.getMongoObjectFactory(mongoDBService), MediaDBFactory.INSTANCE.getMediaDB(mongoDBService), EmptyWindStore.INSTANCE, EmptyGPSFixStore.INSTANCE);
|
||||
}
|
||||
|
||||
public void performReplicationSetup() throws Exception {
|
||||
private void performReplicationSetup() throws Exception {
|
||||
try {
|
||||
replicationDescriptorPair = basicSetUp(false, this.master, this.replica);
|
||||
replicationDescriptorPair = basicSetUp(/* dropDB */ false, this.master, this.replica);
|
||||
} catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
tearDown();
|
||||
@@ -81,7 +81,6 @@ public class InitialLoadReplicationObjectIdentityTest extends AbstractServerRepl
|
||||
|
||||
@Test
|
||||
public void testInitialLoad() throws Exception {
|
||||
|
||||
/* Event */
|
||||
String eventName = "Monster Event";
|
||||
String venue = "Default Venue";
|
||||
|
||||
+6
-7
@@ -10,14 +10,14 @@ import org.junit.Test;
|
||||
|
||||
import com.sap.sailing.domain.leaderboard.Leaderboard;
|
||||
import com.sap.sailing.domain.leaderboard.impl.LowPoint;
|
||||
import com.sap.sailing.server.RacingEventService;
|
||||
import com.sap.sailing.server.operationaltransformation.CreateFlexibleLeaderboard;
|
||||
import com.sap.sse.common.Util;
|
||||
import com.sap.sse.common.Util.Pair;
|
||||
import com.sap.sse.replication.ReplicationMasterDescriptor;
|
||||
import com.sap.sse.replication.impl.Replicator;
|
||||
|
||||
public class PrematureOperationReceiptTest extends AbstractServerReplicationTest {
|
||||
private Replicator replicator;
|
||||
private ReplicationServiceTestImpl replicationService;
|
||||
|
||||
/**
|
||||
* Drops the test DB. Sets up master and replica, starts the JMS message broker and registers the replica with the master.
|
||||
@@ -25,10 +25,9 @@ public class PrematureOperationReceiptTest extends AbstractServerReplicationTest
|
||||
@Before
|
||||
@Override
|
||||
public void setUp() throws Exception {
|
||||
Util.Pair<ReplicationServiceTestImpl, ReplicationMasterDescriptor> result = basicSetUp(true, /* master=null means create a new one */ null,
|
||||
/* replica=null means create a new one */ null);
|
||||
replicationService = result.getA();
|
||||
replicator = replicationService.startToReplicateFromButDontYetFetchInitialLoad(result.getB(), /* startReplicatorSuspended */ true);
|
||||
Pair<com.sap.sse.replication.testsupport.AbstractServerReplicationTest.ReplicationServiceTestImpl<RacingEventService>, ReplicationMasterDescriptor> result =
|
||||
basicSetUp(/* dropDB */ true, /* master=null means create a new one */ null, /* replica=null means create a new one */ null);
|
||||
replicator = replicaReplicator.startToReplicateFromButDontYetFetchInitialLoad(result.getB(), /* startReplicatorSuspended */ true);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -45,7 +44,7 @@ public class PrematureOperationReceiptTest extends AbstractServerReplicationTest
|
||||
final int[] discardThresholds = new int[] { 17, 23 };
|
||||
CreateFlexibleLeaderboard createTestLeaderboard = new CreateFlexibleLeaderboard(leaderboardName, null, discardThresholds, new LowPoint(), null);
|
||||
assertNull(master.getLeaderboardByName(leaderboardName));
|
||||
replicationService.initialLoad(); // serialize the master state before the operation has been applied
|
||||
replicaReplicator.initialLoad(); // serialize the master state before the operation has been applied
|
||||
master.apply(createTestLeaderboard);
|
||||
final Leaderboard masterLeaderboard = master.getLeaderboardByName(leaderboardName);
|
||||
assertNotNull(masterLeaderboard);
|
||||
|
||||
+6
-4
@@ -50,6 +50,7 @@ import com.sap.sailing.domain.racelog.RaceLogEventFactory;
|
||||
import com.sap.sailing.domain.racelog.RaceLogIdentifier;
|
||||
import com.sap.sailing.domain.racelog.RaceLogRaceStatusEvent;
|
||||
import com.sap.sailing.domain.racelog.impl.RaceLogEventAuthorImpl;
|
||||
import com.sap.sailing.server.RacingEventService;
|
||||
import com.sap.sailing.server.impl.RacingEventServiceImpl;
|
||||
import com.sap.sailing.server.operationaltransformation.AddColumnToLeaderboard;
|
||||
import com.sap.sailing.server.operationaltransformation.AddColumnToSeries;
|
||||
@@ -58,25 +59,26 @@ import com.sap.sailing.server.operationaltransformation.CreateFlexibleLeaderboar
|
||||
import com.sap.sailing.server.operationaltransformation.CreateRegattaLeaderboard;
|
||||
import com.sap.sailing.server.operationaltransformation.RenameLeaderboard;
|
||||
import com.sap.sse.common.Util;
|
||||
import com.sap.sse.common.Util.Pair;
|
||||
import com.sap.sse.common.impl.MillisecondsTimePoint;
|
||||
import com.sap.sse.mongodb.MongoDBService;
|
||||
import com.sap.sse.replication.ReplicationMasterDescriptor;
|
||||
|
||||
public class RaceLogReplicationTest extends AbstractServerReplicationTest {
|
||||
private static final String BOAT_CLASS_NAME_49er = "49er";
|
||||
|
||||
private com.sap.sse.common.Util.Pair<ReplicationServiceTestImpl, ReplicationMasterDescriptor> replicationDescriptorPair;
|
||||
private Pair<com.sap.sse.replication.testsupport.AbstractServerReplicationTest.ReplicationServiceTestImpl<RacingEventService>, ReplicationMasterDescriptor> replicationDescriptorPair;
|
||||
|
||||
private RaceLogEvent raceLogEvent;
|
||||
private RaceLogEvent anotherRaceLogEvent;
|
||||
private RaceLogEventAuthor author = new RaceLogEventAuthorImpl("Test Author", 1);
|
||||
|
||||
@Before
|
||||
@Override
|
||||
public void setUp() throws Exception {
|
||||
raceLogEvent = RaceLogEventFactory.INSTANCE.createRaceStatusEvent(new MillisecondsTimePoint(1), author, 42, RaceLogRaceStatus.UNKNOWN);
|
||||
anotherRaceLogEvent = RaceLogEventFactory.INSTANCE.createRaceStatusEvent(new MillisecondsTimePoint(2), author, 42, RaceLogRaceStatus.UNKNOWN);
|
||||
try {
|
||||
replicationDescriptorPair = basicSetUp(true, null, null);
|
||||
replicationDescriptorPair = basicSetUp(/* dropDB */ true, null, null);
|
||||
} catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
tearDown();
|
||||
@@ -265,7 +267,7 @@ public class RaceLogReplicationTest extends AbstractServerReplicationTest {
|
||||
private void addEventToDB(RaceLogIdentifier raceLogIdentifier, RaceLogRaceStatusEvent createRaceStatusEvent, String regattaName, String raceColumnName, String fleetName) {
|
||||
final MongoObjectFactory defaultMongoObjectFactory = PersistenceFactory.INSTANCE.getDefaultMongoObjectFactory();
|
||||
defaultMongoObjectFactory.getDatabase().getLastError(); // wait for regatta write operation to complete
|
||||
final RacingEventServiceImpl temporaryMaster = createNewMaster(MongoDBService.INSTANCE, defaultMongoObjectFactory);
|
||||
final RacingEventServiceImpl temporaryMaster = createNewMaster();
|
||||
Regatta regatta = temporaryMaster.getRegatta(new RegattaName(regattaName+" ("+BOAT_CLASS_NAME_49er+")"));
|
||||
Series series = regatta.getSeries().iterator().next();
|
||||
RaceLog masterLog = series.getRaceColumnByName(raceColumnName).getRaceLog(series.getFleetByName(fleetName));
|
||||
|
||||
+7
@@ -0,0 +1,7 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<classpath>
|
||||
<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER/org.eclipse.jdt.internal.debug.ui.launcher.StandardVMType/JavaSE-1.8"/>
|
||||
<classpathentry kind="con" path="org.eclipse.pde.core.requiredPlugins"/>
|
||||
<classpathentry kind="src" path="src"/>
|
||||
<classpathentry kind="output" path="bin"/>
|
||||
</classpath>
|
||||
+28
@@ -0,0 +1,28 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<projectDescription>
|
||||
<name>com.sap.sse.replication.testsupport</name>
|
||||
<comment></comment>
|
||||
<projects>
|
||||
</projects>
|
||||
<buildSpec>
|
||||
<buildCommand>
|
||||
<name>org.eclipse.jdt.core.javabuilder</name>
|
||||
<arguments>
|
||||
</arguments>
|
||||
</buildCommand>
|
||||
<buildCommand>
|
||||
<name>org.eclipse.pde.ManifestBuilder</name>
|
||||
<arguments>
|
||||
</arguments>
|
||||
</buildCommand>
|
||||
<buildCommand>
|
||||
<name>org.eclipse.pde.SchemaBuilder</name>
|
||||
<arguments>
|
||||
</arguments>
|
||||
</buildCommand>
|
||||
</buildSpec>
|
||||
<natures>
|
||||
<nature>org.eclipse.pde.PluginNature</nature>
|
||||
<nature>org.eclipse.jdt.core.javanature</nature>
|
||||
</natures>
|
||||
</projectDescription>
|
||||
+7
@@ -0,0 +1,7 @@
|
||||
eclipse.preferences.version=1
|
||||
org.eclipse.jdt.core.compiler.codegen.inlineJsrBytecode=enabled
|
||||
org.eclipse.jdt.core.compiler.codegen.targetPlatform=1.8
|
||||
org.eclipse.jdt.core.compiler.compliance=1.8
|
||||
org.eclipse.jdt.core.compiler.problem.assertIdentifier=error
|
||||
org.eclipse.jdt.core.compiler.problem.enumIdentifier=error
|
||||
org.eclipse.jdt.core.compiler.source=1.8
|
||||
+3
@@ -0,0 +1,3 @@
|
||||
eclipse.preferences.version=1
|
||||
pluginProject.extensions=false
|
||||
resolve.requirebundle=false
|
||||
@@ -0,0 +1,12 @@
|
||||
Manifest-Version: 1.0
|
||||
Bundle-ManifestVersion: 2
|
||||
Bundle-Name: Testsupport
|
||||
Bundle-SymbolicName: com.sap.sse.replication.testsupport
|
||||
Bundle-Version: 1.0.0.qualifier
|
||||
Bundle-Vendor: SAP
|
||||
Bundle-RequiredExecutionEnvironment: JavaSE-1.8
|
||||
Require-Bundle: org.junit4;bundle-version="4.8.2",
|
||||
com.sap.sse.replication,
|
||||
com.sap.sse.common,
|
||||
com.rabbitmq.client;bundle-version="2.8.4"
|
||||
Export-Package: com.sap.sse.replication.testsupport
|
||||
@@ -0,0 +1,4 @@
|
||||
source.. = src/
|
||||
output.. = bin/
|
||||
bin.includes = META-INF/,\
|
||||
.
|
||||
+293
@@ -0,0 +1,293 @@
|
||||
package com.sap.sse.replication.testsupport;
|
||||
|
||||
import static org.junit.Assert.assertFalse;
|
||||
|
||||
import java.io.BufferedReader;
|
||||
import java.io.IOException;
|
||||
import java.io.InputStreamReader;
|
||||
import java.io.ObjectInputStream;
|
||||
import java.io.ObjectOutputStream;
|
||||
import java.io.OutputStreamWriter;
|
||||
import java.io.PipedInputStream;
|
||||
import java.io.PipedOutputStream;
|
||||
import java.io.PrintWriter;
|
||||
import java.net.ConnectException;
|
||||
import java.net.InetAddress;
|
||||
import java.net.ServerSocket;
|
||||
import java.net.Socket;
|
||||
import java.net.URL;
|
||||
import java.net.URLConnection;
|
||||
import java.util.UUID;
|
||||
import java.util.zip.GZIPOutputStream;
|
||||
|
||||
import org.junit.After;
|
||||
import org.junit.Before;
|
||||
import org.junit.Rule;
|
||||
import org.junit.rules.Timeout;
|
||||
|
||||
import com.rabbitmq.client.Channel;
|
||||
import com.rabbitmq.client.QueueingConsumer;
|
||||
import com.sap.sse.common.Util.Pair;
|
||||
import com.sap.sse.replication.Replicable;
|
||||
import com.sap.sse.replication.ReplicationMasterDescriptor;
|
||||
import com.sap.sse.replication.ReplicationService;
|
||||
import com.sap.sse.replication.impl.Activator;
|
||||
import com.sap.sse.replication.impl.RabbitOutputStream;
|
||||
import com.sap.sse.replication.impl.ReplicaDescriptor;
|
||||
import com.sap.sse.replication.impl.ReplicationInstancesManager;
|
||||
import com.sap.sse.replication.impl.ReplicationMasterDescriptorImpl;
|
||||
import com.sap.sse.replication.impl.ReplicationServiceImpl;
|
||||
import com.sap.sse.replication.impl.Replicator;
|
||||
import com.sap.sse.replication.impl.SingletonReplicablesProvider;
|
||||
|
||||
public abstract class AbstractServerReplicationTest<ReplicableInterface extends Replicable<?, ?>, ReplicableImpl extends ReplicableInterface> {
|
||||
protected static final int SERVLET_PORT = 9990;
|
||||
protected ReplicableImpl replica;
|
||||
protected ReplicableImpl master;
|
||||
protected ReplicationServiceTestImpl<ReplicableInterface> replicaReplicator;
|
||||
private ReplicaDescriptor replicaDescriptor;
|
||||
private ReplicationServiceImpl masterReplicator;
|
||||
protected ReplicationMasterDescriptor masterDescriptor;
|
||||
|
||||
@Rule public Timeout AbstractTracTracLiveTestTimeout = new Timeout(5 * 60 * 1000); // timeout after 5 minutes
|
||||
private Thread initialLoadTestServerThread;
|
||||
|
||||
/**
|
||||
* Sets up master and replica, starts the JMS message broker and registers the replica with the master. If you want
|
||||
* to drop the DB in your particular test case first, override {@link #persistenceSetUp(boolean)}.
|
||||
*/
|
||||
@Before
|
||||
public void setUp() throws Exception {
|
||||
try {
|
||||
Pair<ReplicationServiceTestImpl<ReplicableInterface>, ReplicationMasterDescriptor> result = basicSetUp(
|
||||
/* dropDB */true, /* master=null means create a new one */null,
|
||||
/* replica=null means create a new one */null);
|
||||
result.getA().startToReplicateFrom(result.getB());
|
||||
} catch (Exception e) {
|
||||
tearDown();
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Gives subclasses a hook to set up a persistence component, such as a DB. This implementation does nothing.
|
||||
*
|
||||
* @param dropDB if <code>true</code> the contents of the persistence layers are to be cleared
|
||||
*/
|
||||
protected void persistenceSetUp(boolean dropDB) {
|
||||
}
|
||||
|
||||
/**
|
||||
* Calls {@link #persistenceSetUp(boolean)} first.
|
||||
* @param master
|
||||
* if not <code>null</code>, the value will be used for {@link #master}; otherwise, a new racing event
|
||||
* service will be created as master
|
||||
* @param replica
|
||||
* if not <code>null</code>, the value will be used for {@link #replica}; otherwise, a new racing event
|
||||
* service will be created as replica
|
||||
*/
|
||||
protected Pair<ReplicationServiceTestImpl<ReplicableInterface>, ReplicationMasterDescriptor> basicSetUp(boolean dropDB,
|
||||
ReplicableImpl master, ReplicableImpl replica) throws IOException, InterruptedException {
|
||||
persistenceSetUp(dropDB);
|
||||
String exchangeName = "test-sapsailinganalytics-exchange";
|
||||
String exchangeHost = "localhost";
|
||||
if (System.getenv(Activator.ENV_VAR_NAME_REPLICATION_HOST) != null) {
|
||||
exchangeHost = System.getenv(Activator.ENV_VAR_NAME_REPLICATION_HOST);
|
||||
}
|
||||
final UUID serverUuid = UUID.randomUUID();
|
||||
if (master != null) {
|
||||
this.master = master;
|
||||
} else {
|
||||
this.master = createNewMaster();
|
||||
}
|
||||
if (replica != null) {
|
||||
this.replica = replica;
|
||||
} else {
|
||||
this.replica = createNewReplica();
|
||||
}
|
||||
ReplicationInstancesManager rim = new ReplicationInstancesManager();
|
||||
masterReplicator = new ReplicationServiceImpl(exchangeName, exchangeHost, 0, rim, new SingletonReplicablesProvider(this.master));
|
||||
replicaDescriptor = new ReplicaDescriptor(InetAddress.getLocalHost(), serverUuid, "");
|
||||
masterReplicator.registerReplica(replicaDescriptor);
|
||||
// connect to exchange host and local server running as master
|
||||
// master server and exchange host can be two different hosts
|
||||
masterDescriptor = new ReplicationMasterDescriptorImpl(exchangeHost, exchangeName, 0, UUID.randomUUID().toString(), "localhost", SERVLET_PORT);
|
||||
ReplicationServiceTestImpl<ReplicableInterface> replicaReplicator = new ReplicationServiceTestImpl<ReplicableInterface>(exchangeName, exchangeHost, rim, replicaDescriptor,
|
||||
this.replica, this.master, masterReplicator, masterDescriptor);
|
||||
Pair<ReplicationServiceTestImpl<ReplicableInterface>, ReplicationMasterDescriptor> result = new Pair<>(replicaReplicator, masterDescriptor);
|
||||
initialLoadTestServerThread = replicaReplicator.startInitialLoadTransmissionServlet();
|
||||
this.replicaReplicator = replicaReplicator;
|
||||
return result;
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a new master replicable. This method is called only after {@link #persistenceSetUp(boolean)} has been called. Therefore,
|
||||
* any initialization that {@link #persistenceSetUp(boolean)} may have performed can be assumed to have taken place in the implementations
|
||||
* of this method.
|
||||
*/
|
||||
abstract protected ReplicableImpl createNewMaster();
|
||||
|
||||
/**
|
||||
* Creates a new replica replicable instance. This method is called only after {@link #persistenceSetUp(boolean)} has been called. Therefore,
|
||||
* any initialization that {@link #persistenceSetUp(boolean)} may have performed can be assumed to have taken place in the implementations
|
||||
* of this method.
|
||||
*/
|
||||
abstract protected ReplicableImpl createNewReplica();
|
||||
|
||||
@After
|
||||
public void tearDown() throws Exception {
|
||||
persistenceTearDown();
|
||||
if (masterReplicator != null) {
|
||||
masterReplicator.unregisterReplica(replicaDescriptor);
|
||||
}
|
||||
if (masterDescriptor != null) {
|
||||
masterDescriptor.stopConnection();
|
||||
}
|
||||
try {
|
||||
if (initialLoadTestServerThread != null) {
|
||||
URLConnection urlConnection = new URL("http://localhost:"+SERVLET_PORT+"/STOP").openConnection(); // stop the initial load test server thread
|
||||
urlConnection.getInputStream().close();
|
||||
initialLoadTestServerThread.join(10000 /* wait 10s */);
|
||||
assertFalse("Expected initial load test server thread to die", initialLoadTestServerThread.isAlive());
|
||||
}
|
||||
} catch (ConnectException ex) {
|
||||
/* do not make tests fail because of a server that has been shut down
|
||||
* or when an exception occured (see setUp()) - let the
|
||||
* original exception propagate */
|
||||
ex.printStackTrace();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Gives subclasses a hook to tear down a persistence component, such as a DB. This implementation does nothing.
|
||||
*/
|
||||
protected void persistenceTearDown() {
|
||||
}
|
||||
|
||||
public void stopReplicatingToMaster() throws IOException {
|
||||
replicaReplicator.stopToReplicateFromMaster();
|
||||
}
|
||||
|
||||
public static class ReplicationServiceTestImpl<ReplicableInterface extends Replicable<?, ?>> extends ReplicationServiceImpl {
|
||||
protected static final int INITIAL_LOAD_PACKAGE_SIZE = 1024*1024;
|
||||
private final ReplicableInterface master;
|
||||
private final ReplicaDescriptor replicaDescriptor;
|
||||
private final ReplicationService masterReplicationService;
|
||||
private final ReplicationMasterDescriptor masterDescriptor;
|
||||
|
||||
public ReplicationServiceTestImpl(String exchangeName, String exchangeHost, ReplicationInstancesManager replicationInstancesManager,
|
||||
ReplicaDescriptor replicaDescriptor, ReplicableInterface replica,
|
||||
ReplicableInterface master, ReplicationService masterReplicationService, ReplicationMasterDescriptor masterDescriptor)
|
||||
throws IOException {
|
||||
super(exchangeName, exchangeHost, 0, replicationInstancesManager, new SingletonReplicablesProvider(replica));
|
||||
this.replicaDescriptor = replicaDescriptor;
|
||||
this.master = master;
|
||||
this.masterReplicationService = masterReplicationService;
|
||||
this.masterDescriptor = masterDescriptor;
|
||||
}
|
||||
|
||||
private Thread startInitialLoadTransmissionServlet() throws InterruptedException {
|
||||
final boolean[] listening = new boolean[] { false };
|
||||
Thread initialLoadTestServerThread = new Thread("Replication initial load test server") {
|
||||
public void run() {
|
||||
ServerSocket ss;
|
||||
try {
|
||||
ss = new ServerSocket(SERVLET_PORT);
|
||||
synchronized (listening) {
|
||||
listening[0] = true;
|
||||
listening.notifyAll();
|
||||
}
|
||||
boolean stop = false;
|
||||
while (!stop) {
|
||||
Socket s = ss.accept();
|
||||
String request = new BufferedReader(new InputStreamReader(s.getInputStream())).readLine();
|
||||
PrintWriter pw = new PrintWriter(new OutputStreamWriter(s.getOutputStream()));
|
||||
pw.println("HTTP/1.1 200 OK");
|
||||
pw.println("Content-Type: text/plain");
|
||||
pw.println();
|
||||
pw.flush();
|
||||
if (request.contains("DEREGISTER")) {
|
||||
// assuming that it is safe to unregister all replicas for tests
|
||||
for (ReplicaDescriptor descriptor : getReplicaInfo()) {
|
||||
unregisterReplica(descriptor);
|
||||
}
|
||||
} else if (request.contains("REGISTER")) {
|
||||
final String uuid = UUID.randomUUID().toString();
|
||||
registerReplicaUuidForMaster(uuid, masterDescriptor);
|
||||
pw.print(uuid.getBytes());
|
||||
} else if (request.contains("INITIAL_LOAD")) {
|
||||
Channel channel = masterReplicationService.createMasterChannel();
|
||||
RabbitOutputStream ros = new RabbitOutputStream(INITIAL_LOAD_PACKAGE_SIZE, channel,
|
||||
/* queueName */ "initial-load-for-TestClient-"+UUID.randomUUID(), /* syncAfterTimeout */ false);
|
||||
pw.println(ros.getQueueName());
|
||||
final GZIPOutputStream gzipOutputStream = new GZIPOutputStream(ros);
|
||||
master.serializeForInitialReplication(gzipOutputStream);
|
||||
gzipOutputStream.finish();
|
||||
ros.close();
|
||||
} else if (request.contains("STOP")) {
|
||||
stop = true;
|
||||
}
|
||||
pw.close();
|
||||
s.close();
|
||||
}
|
||||
ss.close();
|
||||
} catch (IOException e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
};
|
||||
initialLoadTestServerThread.start();
|
||||
synchronized (listening) {
|
||||
while (!listening[0]) {
|
||||
listening.wait();
|
||||
}
|
||||
}
|
||||
return initialLoadTestServerThread;
|
||||
}
|
||||
|
||||
public Replicator startToReplicateFromButDontYetFetchInitialLoad(ReplicationMasterDescriptor master, boolean startReplicatorSuspended)
|
||||
throws IOException {
|
||||
masterReplicationService.registerReplica(replicaDescriptor);
|
||||
registerReplicaUuidForMaster(replicaDescriptor.getUuid().toString(), master);
|
||||
QueueingConsumer consumer = master.getConsumer();
|
||||
final Replicator replicator = new Replicator(master, getReplicablesProvider(), startReplicatorSuspended, consumer);
|
||||
new Thread(replicator).start();
|
||||
return replicator;
|
||||
}
|
||||
|
||||
/**
|
||||
* Clones the {@link #master}'s state to the {@link #replica} using
|
||||
* {@link RacingEventServiceImpl#serializeForInitialReplication(ObjectOutputStream)} and
|
||||
* {@link RacingEventServiceImpl#initiallyFillFrom(ObjectInputStream)} through a piped input/output stream.
|
||||
* @throws InterruptedException
|
||||
*/
|
||||
public void initialLoad() throws IOException, ClassNotFoundException, InterruptedException {
|
||||
PipedOutputStream pos = new PipedOutputStream();
|
||||
PipedInputStream pis = new PipedInputStream(pos);
|
||||
new Thread("clone writer") {
|
||||
public void run() {
|
||||
try {
|
||||
master.serializeForInitialReplication(pos);
|
||||
pos.close();
|
||||
} catch (IOException e) {
|
||||
e.printStackTrace();
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
}.start();
|
||||
for (Replicable<?, ?> replicable : getReplicablesProvider().getReplicables()) {
|
||||
replicable.clearReplicaState();
|
||||
replicable.initiallyFillFrom(pis);
|
||||
}
|
||||
pis.close();
|
||||
}
|
||||
|
||||
public void waitUntilQueueIsEmpty() throws InterruptedException, IllegalAccessException {
|
||||
synchronized (getReplicator()) {
|
||||
while (!getReplicator().isQueueEmpty()) {
|
||||
getReplicator().wait();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<classpath>
|
||||
<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER/org.eclipse.jdt.internal.debug.ui.launcher.StandardVMType/JavaSE-1.8"/>
|
||||
<classpathentry kind="con" path="org.eclipse.pde.core.requiredPlugins"/>
|
||||
<classpathentry kind="src" path="src"/>
|
||||
<classpathentry kind="output" path="bin"/>
|
||||
</classpath>
|
||||
+28
@@ -0,0 +1,28 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<projectDescription>
|
||||
<name>com.sap.sse.security.replication.test</name>
|
||||
<comment></comment>
|
||||
<projects>
|
||||
</projects>
|
||||
<buildSpec>
|
||||
<buildCommand>
|
||||
<name>org.eclipse.jdt.core.javabuilder</name>
|
||||
<arguments>
|
||||
</arguments>
|
||||
</buildCommand>
|
||||
<buildCommand>
|
||||
<name>org.eclipse.pde.ManifestBuilder</name>
|
||||
<arguments>
|
||||
</arguments>
|
||||
</buildCommand>
|
||||
<buildCommand>
|
||||
<name>org.eclipse.pde.SchemaBuilder</name>
|
||||
<arguments>
|
||||
</arguments>
|
||||
</buildCommand>
|
||||
</buildSpec>
|
||||
<natures>
|
||||
<nature>org.eclipse.pde.PluginNature</nature>
|
||||
<nature>org.eclipse.jdt.core.javanature</nature>
|
||||
</natures>
|
||||
</projectDescription>
|
||||
+7
@@ -0,0 +1,7 @@
|
||||
eclipse.preferences.version=1
|
||||
org.eclipse.jdt.core.compiler.codegen.inlineJsrBytecode=enabled
|
||||
org.eclipse.jdt.core.compiler.codegen.targetPlatform=1.8
|
||||
org.eclipse.jdt.core.compiler.compliance=1.8
|
||||
org.eclipse.jdt.core.compiler.problem.assertIdentifier=error
|
||||
org.eclipse.jdt.core.compiler.problem.enumIdentifier=error
|
||||
org.eclipse.jdt.core.compiler.source=1.8
|
||||
+3
@@ -0,0 +1,3 @@
|
||||
eclipse.preferences.version=1
|
||||
pluginProject.extensions=false
|
||||
resolve.requirebundle=false
|
||||
@@ -0,0 +1,15 @@
|
||||
Manifest-Version: 1.0
|
||||
Bundle-ManifestVersion: 2
|
||||
Bundle-Name: Test
|
||||
Bundle-SymbolicName: com.sap.sse.security.replication.test
|
||||
Bundle-Version: 1.0.0.qualifier
|
||||
Bundle-Vendor: SAP
|
||||
Fragment-Host: com.sap.sse.security
|
||||
Bundle-RequiredExecutionEnvironment: JavaSE-1.8
|
||||
Require-Bundle: org.junit4;bundle-version="4.8.2",
|
||||
com.sap.sse.security.userstore.mongodb,
|
||||
com.sap.sse.mongodb,
|
||||
com.mongodb.driver,
|
||||
com.rabbitmq.client,
|
||||
com.sap.sse.replication.testsupport,
|
||||
com.sap.sse.replication.testsupport
|
||||
@@ -0,0 +1,4 @@
|
||||
source.. = src/
|
||||
output.. = bin/
|
||||
bin.includes = META-INF/,\
|
||||
.
|
||||
+13
@@ -0,0 +1,13 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"
|
||||
xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
<parent>
|
||||
<artifactId>root</artifactId>
|
||||
<groupId>com.sap.sailing</groupId>
|
||||
<version>1.0.0-SNAPSHOT</version>
|
||||
</parent>
|
||||
<artifactId>com.sap.sse.security.replication.test</artifactId>
|
||||
<packaging>eclipse-plugin</packaging>
|
||||
</project>
|
||||
+41
@@ -0,0 +1,41 @@
|
||||
package com.sap.sse.security.replication.test;
|
||||
|
||||
import java.util.Properties;
|
||||
|
||||
import com.sap.sse.mongodb.MongoDBService;
|
||||
import com.sap.sse.replication.testsupport.AbstractServerReplicationTest;
|
||||
import com.sap.sse.security.SecurityService;
|
||||
import com.sap.sse.security.impl.SecurityServiceImpl;
|
||||
import com.sap.sse.security.userstore.mongodb.MongoObjectFactory;
|
||||
import com.sap.sse.security.userstore.mongodb.PersistenceFactory;
|
||||
import com.sap.sse.security.userstore.mongodb.UserStoreImpl;
|
||||
|
||||
public abstract class AbstractSecurityReplicationTest extends AbstractServerReplicationTest<SecurityService, SecurityServiceImpl> {
|
||||
private MongoDBService mongoDBService;
|
||||
private MongoObjectFactory mongoObjectFactory;
|
||||
|
||||
@Override
|
||||
protected void persistenceSetUp(boolean dropDB) {
|
||||
mongoDBService = MongoDBService.INSTANCE;
|
||||
if (dropDB) {
|
||||
mongoDBService.getDB().dropDatabase();
|
||||
}
|
||||
mongoObjectFactory = PersistenceFactory.INSTANCE.getMongoObjectFactory(mongoDBService);
|
||||
mongoObjectFactory.getDatabase().requestStart();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected SecurityServiceImpl createNewMaster() {
|
||||
return new SecurityServiceImpl(new UserStoreImpl(), new Properties());
|
||||
}
|
||||
|
||||
@Override
|
||||
protected SecurityServiceImpl createNewReplica() {
|
||||
return new SecurityServiceImpl(new UserStoreImpl(), new Properties());
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void persistenceTearDown() {
|
||||
mongoObjectFactory.getDatabase().requestDone();
|
||||
}
|
||||
}
|
||||
+5
@@ -0,0 +1,5 @@
|
||||
package com.sap.sse.security.replication.test;
|
||||
|
||||
public class SimpleSecurityReplicationTest extends AbstractSecurityReplicationTest {
|
||||
|
||||
}
|
||||
+3
@@ -2,6 +2,7 @@ package com.sap.sse.security.userstore.mongodb;
|
||||
|
||||
import java.util.Map;
|
||||
|
||||
import com.mongodb.DB;
|
||||
import com.sap.sse.security.User;
|
||||
|
||||
public interface MongoObjectFactory {
|
||||
@@ -21,4 +22,6 @@ public interface MongoObjectFactory {
|
||||
public void clearAllSettings();
|
||||
|
||||
public void clearAllPreferences();
|
||||
|
||||
public DB getDatabase();
|
||||
}
|
||||
+5
@@ -27,6 +27,11 @@ public class MongoObjectFactoryImpl implements MongoObjectFactory {
|
||||
this.db = db;
|
||||
}
|
||||
|
||||
@Override
|
||||
public DB getDatabase() {
|
||||
return db;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void storeUser(User user) {
|
||||
DBCollection usersCollection = db.getCollection(CollectionNames.USERS.name());
|
||||
|
||||
@@ -24,6 +24,7 @@
|
||||
<module>com.sap.sse.datamining.shared</module>
|
||||
<module>com.sap.sse.datamining.test</module>
|
||||
<module>com.sap.sse.security</module>
|
||||
<module>com.sap.sse.security.replication.test</module>
|
||||
<module>com.sap.sse.security.test</module>
|
||||
<module>com.sap.sse.security.ui</module>
|
||||
<module>com.sap.sse.security.userstore.mongodb</module>
|
||||
|
||||
Reference in new issue
Block a user