001: /*
002: * All content copyright (c) 2003-2007 Terracotta, Inc., except as may otherwise be noted in a separate copyright
003: * notice. All rights reserved.
004: */
005: package com.tc.l2.objectserver;
006:
007: import com.tc.async.impl.MockSink;
008: import com.tc.async.impl.OrderedSink;
009: import com.tc.logging.TCLogging;
010: import com.tc.net.groups.ClientID;
011: import com.tc.net.groups.SingleNodeGroupManager;
012: import com.tc.net.protocol.tcm.ChannelID;
013: import com.tc.object.ObjectID;
014: import com.tc.object.dmi.DmiDescriptor;
015: import com.tc.object.dna.api.DNA;
016: import com.tc.object.dna.impl.ObjectStringSerializer;
017: import com.tc.object.gtx.GlobalTransactionID;
018: import com.tc.object.lockmanager.api.LockID;
019: import com.tc.object.tx.TransactionID;
020: import com.tc.object.tx.TxnBatchID;
021: import com.tc.object.tx.TxnType;
022: import com.tc.objectserver.core.api.TestDNA;
023: import com.tc.objectserver.gtx.TestGlobalTransactionManager;
024: import com.tc.objectserver.tx.ServerTransaction;
025: import com.tc.objectserver.tx.ServerTransactionImpl;
026: import com.tc.objectserver.tx.TestServerTransactionManager;
027: import com.tc.util.SequenceID;
028:
029: import java.util.ArrayList;
030: import java.util.Collection;
031: import java.util.Collections;
032: import java.util.HashMap;
033: import java.util.HashSet;
034: import java.util.Iterator;
035: import java.util.LinkedHashMap;
036: import java.util.LinkedList;
037: import java.util.List;
038: import java.util.Map;
039: import java.util.Set;
040:
041: import junit.framework.TestCase;
042:
043: public class ReplicatedTransactionManagerTest extends TestCase {
044:
045: ReplicatedTransactionManagerImpl rtm;
046: SingleNodeGroupManager grpMgr;
047: TestServerTransactionManager txnMgr;
048: TestGlobalTransactionManager gtxm;
049: ClientID clientID;
050:
051: public void setUp() throws Exception {
052: clientID = new ClientID(new ChannelID(1));
053: grpMgr = new SingleNodeGroupManager();
054: txnMgr = new TestServerTransactionManager();
055: gtxm = new TestGlobalTransactionManager();
056: rtm = new ReplicatedTransactionManagerImpl(
057: grpMgr,
058: new OrderedSink(
059: TCLogging
060: .getLogger(ReplicatedTransactionManagerTest.class),
061: new MockSink()), txnMgr, gtxm);
062: }
063:
064: /**
065: * Some basic tests, a zillion other scenarios could be tested if only I had time :(
066: */
067: public void testPassiveUninitialized() throws Exception {
068:
069: Set knownIds = new HashSet();
070: knownIds.add(new ObjectID(1));
071: knownIds.add(new ObjectID(2));
072:
073: // two objects are already present
074: rtm.init(knownIds);
075:
076: LinkedHashMap txns = createTxns(1, 1, 2, false);
077: rtm.addCommitedTransactions(clientID, txns.keySet(), txns
078: .values());
079:
080: // Since both are know oids, transactions should pass thru
081: assertAndClear(txns.values());
082:
083: // create a txn containing a new Object (OID 3)
084: txns = createTxns(1, 3, 1, true);
085: rtm.addCommitedTransactions(clientID, txns.keySet(), txns
086: .values());
087:
088: // Should go thru too
089: assertAndClear(txns.values());
090:
091: // Now create a txn with all three objects
092: txns = createTxns(1, 1, 3, false);
093: rtm.addCommitedTransactions(clientID, txns.keySet(), txns
094: .values());
095:
096: // Since all are known oids, transactions should pass thru
097: assertAndClear(txns.values());
098:
099: // Now create a txn with all unknown ObjectIDs (4,5,6)
100: txns = createTxns(1, 4, 3, false);
101: rtm.addCommitedTransactions(clientID, txns.keySet(), txns
102: .values());
103:
104: // None should be sent thru
105: assertTrue(txnMgr.incomingTxns.isEmpty());
106:
107: // Create more txns with all unknown ObjectIDs (7,8,9)
108: LinkedHashMap txns1 = createTxns(1, 7, 1, false);
109: rtm.addCommitedTransactions(clientID, txns1.keySet(), txns1
110: .values());
111: LinkedHashMap txns2 = createTxns(1, 8, 2, false);
112: rtm.addCommitedTransactions(clientID, txns2.keySet(), txns2
113: .values());
114:
115: // None should be sent thru
116: assertTrue(txnMgr.incomingTxns.isEmpty());
117:
118: // Now create Object Sync Txn for 4,5,6
119: LinkedHashMap syncTxns = createTxns(1, 4, 3, true);
120: rtm.addObjectSyncTransaction((ServerTransaction) syncTxns
121: .values().iterator().next());
122:
123: // One Compound Transaction containing the object DNA and the delta DNA should be sent to the
124: // transactionalObjectManager
125: assertTrue(txnMgr.incomingTxns.size() == 1);
126: ServerTransaction gotTxn = (ServerTransaction) txnMgr.incomingTxns
127: .remove(0);
128: assertContainsAllAndRemove((ServerTransaction) syncTxns
129: .values().iterator().next(), gotTxn);
130: assertContainsAllVersionizedAndRemove((ServerTransaction) txns
131: .values().iterator().next(), gotTxn);
132: assertTrue(gotTxn.getChanges().isEmpty());
133:
134: // Now send transaction complete for txn1, with new Objects (10), this should clear pending changes for 7
135: txns = createTxns(1, 10, 1, true);
136: rtm.addCommitedTransactions(clientID, txns.keySet(), txns
137: .values());
138: rtm
139: .clearTransactionsBelowLowWaterMark(getNextLowWaterMark(txns1
140: .values()));
141: assertAndClear(txns.values());
142:
143: // Now create Object Sync txn for 7,8,9
144: syncTxns = createTxns(1, 7, 3, true);
145: rtm.addObjectSyncTransaction((ServerTransaction) syncTxns
146: .values().iterator().next());
147:
148: // One Compound Transaction containing the object DNA for 7 and object DNA and the delta DNA for 8,9 should be sent
149: // to the transactionalObjectManager
150: assertTrue(txnMgr.incomingTxns.size() == 1);
151: gotTxn = (ServerTransaction) txnMgr.incomingTxns.remove(0);
152: List changes = gotTxn.getChanges();
153: assertEquals(5, changes.size());
154: DNA dna = (DNA) changes.get(0);
155: assertEquals(new ObjectID(7), dna.getObjectID());
156: assertFalse(dna.isDelta()); // New object
157: dna = (DNA) changes.get(1);
158: assertEquals(new ObjectID(8), dna.getObjectID());
159: assertFalse(dna.isDelta()); // New object
160: dna = (DNA) changes.get(2);
161: assertEquals(new ObjectID(8), dna.getObjectID());
162: assertTrue(dna.isDelta()); // Change to that object
163: dna = (DNA) changes.get(3);
164: assertEquals(new ObjectID(9), dna.getObjectID());
165: assertFalse(dna.isDelta()); // New object
166: dna = (DNA) changes.get(4);
167: assertEquals(new ObjectID(9), dna.getObjectID());
168: assertTrue(dna.isDelta()); // Change to that object
169: }
170:
171: private GlobalTransactionID getNextLowWaterMark(Collection txns) {
172: GlobalTransactionID lwm = GlobalTransactionID.NULL_ID;
173: for (Iterator i = txns.iterator(); i.hasNext();) {
174: ServerTransaction txn = (ServerTransaction) i.next();
175: if (lwm.toLong() < txn.getGlobalTransactionID().toLong()) {
176: lwm = txn.getGlobalTransactionID();
177: }
178: }
179: return lwm.next();
180: }
181:
182: private void assertContainsAllVersionizedAndRemove(
183: ServerTransaction expected, ServerTransaction got) {
184: List c1 = expected.getChanges();
185: List c2 = got.getChanges();
186: assertEquals(c1.size(), c2.size());
187: for (Iterator i = c2.iterator(); i.hasNext();) {
188: DNA dna = (DNA) i.next();
189: assertEquals(expected.getGlobalTransactionID().toLong(),
190: dna.getVersion());
191: boolean found = false;
192: for (Iterator j = c1.iterator(); j.hasNext();) {
193: DNA orgDNA = (DNA) j.next();
194: // XXX:: This depends on the fact that we dont create a resetable cursor when creating a VersionizedDNAWrapper
195: // in ReplicatedTransactionManagerImpl
196: if (dna.getCursor() == orgDNA.getCursor()) {
197: found = true;
198: break;
199: }
200: }
201: assertTrue(found);
202: i.remove();
203: }
204: }
205:
206: private void assertContainsAllAndRemove(ServerTransaction expected,
207: ServerTransaction got) {
208: List c1 = expected.getChanges();
209: List c2 = got.getChanges();
210: for (Iterator i = c1.iterator(); i.hasNext();) {
211: DNA dna = (DNA) i.next();
212: assertTrue(c2.remove(dna));
213: }
214: }
215:
216: private void assertAndClear(Collection txns) {
217: assertEquals(new ArrayList(txns), txnMgr.incomingTxns);
218: txnMgr.incomingTxns.clear();
219: }
220:
221: long bid = 0;
222: long sid = 0;
223: long tid = 0;
224:
225: private LinkedHashMap createTxns(int txnCount, int oidStart,
226: int objectCount, boolean newObjects) {
227: LinkedHashMap map = new LinkedHashMap();
228:
229: TxnBatchID batchID = new TxnBatchID(bid++);
230: LockID[] lockIDs = new LockID[] { new LockID("1") };
231: ObjectStringSerializer serializer = null;
232: Map newRoots = Collections.unmodifiableMap(new HashMap());
233: TxnType txnType = TxnType.NORMAL;
234: List notifies = new LinkedList();
235:
236: for (int i = 0; i < txnCount; i++) {
237: List dnas = new LinkedList();
238: SequenceID sequenceID = new SequenceID(sid++);
239: TransactionID txID = new TransactionID(tid++);
240: for (int j = oidStart; j < oidStart + objectCount; j++) {
241: dnas.add(new TestDNA(new ObjectID(j), !newObjects));
242: }
243: ServerTransaction tx = new ServerTransactionImpl(gtxm,
244: batchID, txID, sequenceID, lockIDs, clientID, dnas,
245: serializer, newRoots, txnType, notifies,
246: DmiDescriptor.EMPTY_ARRAY);
247: map.put(tx.getServerTransactionID(), tx);
248: }
249: return map;
250: }
251:
252: }
|