[hornetq-commits] JBoss hornetq SVN: r8070 - in branches/Replication_Clebert: src/main/org/hornetq/core/paging/impl and 8 other directories.
do-not-reply at jboss.org
do-not-reply at jboss.org
Thu Oct 8 17:40:37 EDT 2009
Author: clebert.suconic at jboss.com
Date: 2009-10-08 17:40:37 -0400 (Thu, 08 Oct 2009)
New Revision: 8070
Added:
branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/ReplicationPageEventMessage.java
branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/ReplicationPageWriteMessage.java
Removed:
branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/ReplicationPacket.java
Modified:
branches/Replication_Clebert/src/main/org/hornetq/core/paging/PagingStore.java
branches/Replication_Clebert/src/main/org/hornetq/core/paging/impl/PageImpl.java
branches/Replication_Clebert/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java
branches/Replication_Clebert/src/main/org/hornetq/core/persistence/StorageManager.java
branches/Replication_Clebert/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java
branches/Replication_Clebert/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java
branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/PacketDecoder.java
branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/PacketImpl.java
branches/Replication_Clebert/src/main/org/hornetq/core/replication/ReplicationManager.java
branches/Replication_Clebert/src/main/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java
branches/Replication_Clebert/src/main/org/hornetq/core/replication/impl/ReplicationManagerImpl.java
branches/Replication_Clebert/tests/src/org/hornetq/tests/integration/paging/PageCrashTest.java
Log:
Paging changes
Modified: branches/Replication_Clebert/src/main/org/hornetq/core/paging/PagingStore.java
===================================================================
--- branches/Replication_Clebert/src/main/org/hornetq/core/paging/PagingStore.java 2009-10-08 17:09:12 UTC (rev 8069)
+++ branches/Replication_Clebert/src/main/org/hornetq/core/paging/PagingStore.java 2009-10-08 21:40:37 UTC (rev 8070)
@@ -54,6 +54,8 @@
public boolean readPage() throws Exception;
Page getCurrentPage();
+
+ Page createPage(final int page) throws Exception;
/**
*
Modified: branches/Replication_Clebert/src/main/org/hornetq/core/paging/impl/PageImpl.java
===================================================================
--- branches/Replication_Clebert/src/main/org/hornetq/core/paging/impl/PageImpl.java 2009-10-08 17:09:12 UTC (rev 8069)
+++ branches/Replication_Clebert/src/main/org/hornetq/core/paging/impl/PageImpl.java 2009-10-08 21:40:37 UTC (rev 8070)
@@ -28,6 +28,8 @@
import org.hornetq.core.logging.Logger;
import org.hornetq.core.paging.Page;
import org.hornetq.core.paging.PagedMessage;
+import org.hornetq.core.persistence.StorageManager;
+import org.hornetq.utils.SimpleString;
/**
*
@@ -59,18 +61,30 @@
private final SequentialFileFactory fileFactory;
private final AtomicInteger size = new AtomicInteger(0);
+
+ private final StorageManager storageManager;
+
+ private final SimpleString storeName;
// Static --------------------------------------------------------
// Constructors --------------------------------------------------
- public PageImpl(final SequentialFileFactory factory, final SequentialFile file, final int pageId) throws Exception
+ public PageImpl(final SimpleString storeName, final StorageManager storageManager, final SequentialFileFactory factory, final SequentialFile file, final int pageId) throws Exception
{
this.pageId = pageId;
this.file = file;
- fileFactory = factory;
+ this.fileFactory = factory;
+ this.storageManager = storageManager;
+ this.storeName = storeName;
}
+
+ public PageImpl(final SequentialFileFactory factory, final SequentialFile file, final int pageId) throws Exception
+ {
+ this(null, null, factory, file, pageId);
+ }
+
// Public --------------------------------------------------------
// PagingFile implementation
@@ -154,9 +168,11 @@
numberOfMessages.incrementAndGet();
size.addAndGet(buffer.limit());
+ storageManager.pageWrite(message, pageId);
+
if (message.getMessage(null).isLargeMessage())
{
- // If we don't sync on large messages we could have the risk of files unnatended files on disk
+ // If we don't sync on large messages we could have the risk of unattended files on disk
sync();
}
}
@@ -175,6 +191,7 @@
public void close() throws Exception
{
+ storageManager.pageClosed(storeName, pageId);
file.close();
}
Modified: branches/Replication_Clebert/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java
===================================================================
--- branches/Replication_Clebert/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java 2009-10-08 17:09:12 UTC (rev 8069)
+++ branches/Replication_Clebert/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java 2009-10-08 21:40:37 UTC (rev 8070)
@@ -607,6 +607,35 @@
return currentPage;
}
+
+ public Page createPage(final int page) throws Exception
+ {
+ String fileName = createFileName(page);
+
+ if (fileFactory == null)
+ {
+ fileFactory = storeFactory.newFileFactory(getStoreName());
+ }
+
+ SequentialFile file = fileFactory.createSequentialFile(fileName, 1000);
+
+ file.open();
+
+ long size = file.size();
+
+ if (fileFactory.isSupportsCallbacks() && size < pageSize)
+ {
+ file.fill((int)size, (int)(pageSize - size), (byte)0);
+ }
+
+ file.position(0);
+
+ file.close();
+
+ return new PageImpl(this.storeName, storageManager, fileFactory, file, page);
+ }
+
+
// TestSupportPageStore ------------------------------------------
public void forceAnotherPage() throws Exception
@@ -700,36 +729,6 @@
// Protected -----------------------------------------------------
- // In order to test failures, we need to be able to extend this class
- // and replace the Page for another Page that will fail before the file is removed
- // That's why createPage is not a private method
- protected Page createPage(final int page) throws Exception
- {
- String fileName = createFileName(page);
-
- if (fileFactory == null)
- {
- fileFactory = storeFactory.newFileFactory(getStoreName());
- }
-
- SequentialFile file = fileFactory.createSequentialFile(fileName, 1000);
-
- file.open();
-
- long size = file.size();
-
- if (fileFactory.isSupportsCallbacks() && size < pageSize)
- {
- file.fill((int)size, (int)(pageSize - size), (byte)0);
- }
-
- file.position(0);
-
- file.close();
-
- return new PageImpl(fileFactory, file, page);
- }
-
// Private -------------------------------------------------------
/**
@@ -931,8 +930,9 @@
{
currentPage.close();
}
-
+
currentPage = createPage(currentPageId);
+
currentPageSize.set(0);
Modified: branches/Replication_Clebert/src/main/org/hornetq/core/persistence/StorageManager.java
===================================================================
--- branches/Replication_Clebert/src/main/org/hornetq/core/persistence/StorageManager.java 2009-10-08 17:09:12 UTC (rev 8069)
+++ branches/Replication_Clebert/src/main/org/hornetq/core/persistence/StorageManager.java 2009-10-08 21:40:37 UTC (rev 8070)
@@ -18,7 +18,9 @@
import javax.transaction.xa.Xid;
+import org.hornetq.core.buffers.ChannelBuffer;
import org.hornetq.core.paging.PageTransactionInfo;
+import org.hornetq.core.paging.PagedMessage;
import org.hornetq.core.paging.PagingManager;
import org.hornetq.core.postoffice.Binding;
import org.hornetq.core.server.HornetQComponent;
@@ -44,6 +46,12 @@
{
// Message related operations
+ void pageClosed(SimpleString storeName, int pageNumber);
+
+ void pageDeleted(SimpleString storeName, int pageNumber);
+
+ void pageWrite(PagedMessage message, int pageNumber);
+
boolean isReplicated();
void afterReplicated(Runnable run);
Modified: branches/Replication_Clebert/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java
===================================================================
--- branches/Replication_Clebert/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2009-10-08 17:09:12 UTC (rev 8069)
+++ branches/Replication_Clebert/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2009-10-08 21:40:37 UTC (rev 8070)
@@ -29,6 +29,7 @@
import javax.transaction.xa.Xid;
+import org.hornetq.core.buffers.ChannelBuffer;
import org.hornetq.core.buffers.ChannelBuffers;
import org.hornetq.core.config.Configuration;
import org.hornetq.core.exception.HornetQException;
@@ -47,6 +48,7 @@
import org.hornetq.core.logging.Logger;
import org.hornetq.core.message.impl.MessageImpl;
import org.hornetq.core.paging.PageTransactionInfo;
+import org.hornetq.core.paging.PagedMessage;
import org.hornetq.core.paging.PagingManager;
import org.hornetq.core.paging.impl.PageTransactionInfoImpl;
import org.hornetq.core.persistence.QueueBindingInfo;
@@ -290,6 +292,47 @@
return replicator != null;
}
+
+ // TODO: shouldn't those page methods be on the PageManager?
+
+ /*
+ *
+ * (non-Javadoc)
+ * @see org.hornetq.core.persistence.StorageManager#pageClosed(org.hornetq.utils.SimpleString, int)
+ */
+ public void pageClosed(SimpleString storeName, int pageNumber)
+ {
+ if (isReplicated())
+ {
+ replicator.pageClosed(storeName, pageNumber);
+ }
+ }
+
+ /* (non-Javadoc)
+ * @see org.hornetq.core.persistence.StorageManager#pageDeleted(org.hornetq.utils.SimpleString, int)
+ */
+ public void pageDeleted(SimpleString storeName, int pageNumber)
+ {
+ if (isReplicated())
+ {
+ replicator.pageDeleted(storeName, pageNumber);
+ }
+ }
+
+ /* (non-Javadoc)
+ * @see org.hornetq.core.persistence.StorageManager#pageWrite(org.hornetq.utils.SimpleString, int, org.hornetq.core.buffers.ChannelBuffer)
+ */
+ public void pageWrite(PagedMessage message, int pageNumber)
+ {
+ if (isReplicated())
+ {
+ replicator.pageWrite(message, pageNumber);
+ }
+ }
+
+
+ // TODO: shouldn't those page methods be on the PageManager? ^^^^
+
public void afterReplicated(Runnable run)
{
if (replicator == null)
Modified: branches/Replication_Clebert/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java
===================================================================
--- branches/Replication_Clebert/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java 2009-10-08 17:09:12 UTC (rev 8069)
+++ branches/Replication_Clebert/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java 2009-10-08 21:40:37 UTC (rev 8070)
@@ -21,6 +21,7 @@
import org.hornetq.core.logging.Logger;
import org.hornetq.core.paging.PageTransactionInfo;
+import org.hornetq.core.paging.PagedMessage;
import org.hornetq.core.paging.PagingManager;
import org.hornetq.core.persistence.QueueBindingInfo;
import org.hornetq.core.persistence.StorageManager;
@@ -268,4 +269,25 @@
{
}
+ /* (non-Javadoc)
+ * @see org.hornetq.core.persistence.StorageManager#pageClosed(org.hornetq.utils.SimpleString, int)
+ */
+ public void pageClosed(SimpleString storeName, int pageNumber)
+ {
+ }
+
+ /* (non-Javadoc)
+ * @see org.hornetq.core.persistence.StorageManager#pageDeleted(org.hornetq.utils.SimpleString, int)
+ */
+ public void pageDeleted(SimpleString storeName, int pageNumber)
+ {
+ }
+
+ /* (non-Javadoc)
+ * @see org.hornetq.core.persistence.StorageManager#pageWrite(org.hornetq.core.paging.PagedMessage, int)
+ */
+ public void pageWrite(PagedMessage message, int pageNumber)
+ {
+ }
+
}
Modified: branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/PacketDecoder.java
===================================================================
--- branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/PacketDecoder.java 2009-10-08 17:09:12 UTC (rev 8069)
+++ branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/PacketDecoder.java 2009-10-08 21:40:37 UTC (rev 8070)
@@ -12,6 +12,9 @@
*/
package org.hornetq.core.remoting.impl;
+
+import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATION_PAGE_EVENT;
+import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATION_PAGE_WRITE;
import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATION_PREPARE;
import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATION_DELETE_TX;
import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATION_COMMIT_ROLLBACK;
@@ -405,6 +408,16 @@
packet = new ReplicationResponseMessage();
break;
}
+ case REPLICATION_PAGE_WRITE:
+ {
+ packet = new ReplicationResponseMessage();
+ break;
+ }
+ case REPLICATION_PAGE_EVENT:
+ {
+ packet = new ReplicationResponseMessage();
+ break;
+ }
default:
{
throw new IllegalArgumentException("Invalid type: " + packetType);
Modified: branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/PacketImpl.java
===================================================================
--- branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/PacketImpl.java 2009-10-08 17:09:12 UTC (rev 8069)
+++ branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/PacketImpl.java 2009-10-08 21:40:37 UTC (rev 8070)
@@ -155,7 +155,11 @@
public static final byte REPLICATION_COMMIT_ROLLBACK = 86;
+ public static final byte REPLICATION_PAGE_WRITE = 87;
+
+ public static final byte REPLICATION_PAGE_EVENT = 88;
+
// Static --------------------------------------------------------
Deleted: branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/ReplicationPacket.java
===================================================================
--- branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/ReplicationPacket.java 2009-10-08 17:09:12 UTC (rev 8069)
+++ branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/ReplicationPacket.java 2009-10-08 21:40:37 UTC (rev 8070)
@@ -1,53 +0,0 @@
-/*
- * Copyright 2009 Red Hat, Inc.
- * Red Hat licenses this file to you under the Apache License, version
- * 2.0 (the "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- * http://www.apache.org/licenses/LICENSE-2.0
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
- * implied. See the License for the specific language governing
- * permissions and limitations under the License.
- */
-
-package org.hornetq.core.remoting.impl.wireformat;
-
-/**
- * A ReplicationPacket
- *
- * @author <mailto:clebert.suconic at jboss.org">Clebert Suconic</a>
- *
- *
- */
-public class ReplicationPacket extends PacketImpl
-{
-
- /**
- * @param type
- */
- public ReplicationPacket(byte type)
- {
- super(type);
- // TODO Auto-generated constructor stub
- }
-
- // Constants -----------------------------------------------------
-
- // Attributes ----------------------------------------------------
-
- // Static --------------------------------------------------------
-
- // Constructors --------------------------------------------------
-
- // Public --------------------------------------------------------
-
- // Package protected ---------------------------------------------
-
- // Protected -----------------------------------------------------
-
- // Private -------------------------------------------------------
-
- // Inner classes -------------------------------------------------
-
-}
Added: branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/ReplicationPageEventMessage.java
===================================================================
--- branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/ReplicationPageEventMessage.java (rev 0)
+++ branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/ReplicationPageEventMessage.java 2009-10-08 21:40:37 UTC (rev 8070)
@@ -0,0 +1,117 @@
+/*
+ * Copyright 2009 Red Hat, Inc.
+ * Red Hat licenses this file to you under the Apache License, version
+ * 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ * implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package org.hornetq.core.remoting.impl.wireformat;
+
+import org.hornetq.core.remoting.spi.HornetQBuffer;
+import org.hornetq.utils.DataConstants;
+import org.hornetq.utils.SimpleString;
+
+/**
+ * A ReplicationPageWrite
+ *
+ * @author <mailto:clebert.suconic at jboss.org">Clebert Suconic</a>
+ *
+ *
+ */
+public class ReplicationPageEventMessage extends PacketImpl
+{
+
+ private int pageNumber;
+
+ private SimpleString storeName;
+
+ /**
+ * True = delete page, False = close page
+ */
+ private boolean isDelete;
+
+ // Constants -----------------------------------------------------
+
+ // Attributes ----------------------------------------------------
+
+ // Static --------------------------------------------------------
+
+ // Constructors --------------------------------------------------
+
+ public ReplicationPageEventMessage()
+ {
+ super(REPLICATION_PAGE_EVENT);
+ }
+
+ public ReplicationPageEventMessage(final SimpleString storeName, final int pageNumber, final boolean isDelete)
+ {
+ this();
+ this.pageNumber = pageNumber;
+ this.isDelete = isDelete;
+ this.storeName = storeName;
+ }
+
+ // Public --------------------------------------------------------
+
+ @Override
+ public int getRequiredBufferSize()
+ {
+ return BASIC_PACKET_SIZE + DataConstants.SIZE_INT + storeName.sizeof() + DataConstants.SIZE_BOOLEAN;
+
+ }
+
+ @Override
+ public void encodeBody(final HornetQBuffer buffer)
+ {
+ buffer.writeSimpleString(storeName);
+ buffer.writeInt(pageNumber);
+ buffer.writeBoolean(isDelete);
+ }
+
+ @Override
+ public void decodeBody(final HornetQBuffer buffer)
+ {
+ storeName = buffer.readSimpleString();
+ pageNumber = buffer.readInt();
+ isDelete = buffer.readBoolean();
+ }
+
+ /**
+ * @return the pageNumber
+ */
+ public int getPageNumber()
+ {
+ return pageNumber;
+ }
+
+ /**
+ * @return the storeName
+ */
+ public SimpleString getStoreName()
+ {
+ return storeName;
+ }
+
+ /**
+ * @return the isDelete
+ */
+ public boolean isDelete()
+ {
+ return isDelete;
+ }
+
+ // Package protected ---------------------------------------------
+
+ // Protected -----------------------------------------------------
+
+ // Private -------------------------------------------------------
+
+ // Inner classes -------------------------------------------------
+
+}
Added: branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/ReplicationPageWriteMessage.java
===================================================================
--- branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/ReplicationPageWriteMessage.java (rev 0)
+++ branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/ReplicationPageWriteMessage.java 2009-10-08 21:40:37 UTC (rev 8070)
@@ -0,0 +1,107 @@
+/*
+ * Copyright 2009 Red Hat, Inc.
+ * Red Hat licenses this file to you under the Apache License, version
+ * 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ * implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package org.hornetq.core.remoting.impl.wireformat;
+
+import org.hornetq.core.paging.PagedMessage;
+import org.hornetq.core.paging.impl.PagedMessageImpl;
+import org.hornetq.core.remoting.spi.HornetQBuffer;
+import org.hornetq.utils.DataConstants;
+
+/**
+ * A ReplicationPageWrite
+ *
+ * @author <mailto:clebert.suconic at jboss.org">Clebert Suconic</a>
+ *
+ *
+ */
+public class ReplicationPageWriteMessage extends PacketImpl
+{
+
+ int pageNumber;
+
+ PagedMessage pagedMessage;
+
+ // Constants -----------------------------------------------------
+
+ // Attributes ----------------------------------------------------
+
+ // Static --------------------------------------------------------
+
+ // Constructors --------------------------------------------------
+
+ public ReplicationPageWriteMessage()
+ {
+ super(REPLICATION_PAGE_WRITE);
+ }
+
+ public ReplicationPageWriteMessage(final PagedMessage pagedMessage, final int pageNumber)
+ {
+ this();
+ this.pageNumber = pageNumber;
+ this.pagedMessage = pagedMessage;
+ }
+
+ // Public --------------------------------------------------------
+
+ public int getRequiredBufferSize()
+ {
+ return BASIC_PACKET_SIZE +
+ DataConstants.SIZE_INT +
+ pagedMessage.getEncodeSize();
+
+ }
+
+ @Override
+ public void encodeBody(final HornetQBuffer buffer)
+ {
+ buffer.writeInt(pageNumber);
+ pagedMessage.encode(buffer);
+ }
+
+ @Override
+ public void decodeBody(final HornetQBuffer buffer)
+ {
+ this.pageNumber = buffer.readInt();
+ pagedMessage = new PagedMessageImpl();
+ pagedMessage.decode(buffer);
+ }
+
+ /**
+ * @return the pageNumber
+ */
+ public int getPageNumber()
+ {
+ return pageNumber;
+ }
+
+ /**
+ * @return the pagedMessage
+ */
+ public PagedMessage getPagedMessage()
+ {
+ return pagedMessage;
+ }
+
+
+
+
+ // Package protected ---------------------------------------------
+
+ // Protected -----------------------------------------------------
+
+ // Private -------------------------------------------------------
+
+ // Inner classes -------------------------------------------------
+
+}
Modified: branches/Replication_Clebert/src/main/org/hornetq/core/replication/ReplicationManager.java
===================================================================
--- branches/Replication_Clebert/src/main/org/hornetq/core/replication/ReplicationManager.java 2009-10-08 17:09:12 UTC (rev 8069)
+++ branches/Replication_Clebert/src/main/org/hornetq/core/replication/ReplicationManager.java 2009-10-08 21:40:37 UTC (rev 8070)
@@ -16,7 +16,9 @@
import java.util.Set;
import org.hornetq.core.journal.EncodingSupport;
+import org.hornetq.core.paging.PagedMessage;
import org.hornetq.core.server.HornetQComponent;
+import org.hornetq.utils.SimpleString;
/**
* @author <mailto:clebert.suconic at jboss.org">Clebert Suconic</a>
@@ -53,4 +55,23 @@
/** A list of tokens that are still waiting for replications to be completed */
Set<ReplicationToken> getActiveTokens();
+ /**
+ * @param storeName
+ * @param pageNumber
+ */
+ void pageClosed(SimpleString storeName, int pageNumber);
+
+ /**
+ * @param storeName
+ * @param pageNumber
+ */
+ void pageDeleted(SimpleString storeName, int pageNumber);
+
+
+ /**
+ * @param storeName
+ * @param pageNumber
+ */
+ void pageWrite(PagedMessage message, int pageNumber);
+
}
Modified: branches/Replication_Clebert/src/main/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java
===================================================================
--- branches/Replication_Clebert/src/main/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java 2009-10-08 17:09:12 UTC (rev 8069)
+++ branches/Replication_Clebert/src/main/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java 2009-10-08 21:40:37 UTC (rev 8070)
@@ -13,9 +13,17 @@
package org.hornetq.core.replication.impl;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
import org.hornetq.core.config.Configuration;
import org.hornetq.core.journal.Journal;
import org.hornetq.core.logging.Logger;
+import org.hornetq.core.paging.Page;
+import org.hornetq.core.paging.PagedMessage;
+import org.hornetq.core.paging.PagingManager;
+import org.hornetq.core.paging.impl.PagingManagerImpl;
+import org.hornetq.core.paging.impl.PagingStoreFactoryNIO;
import org.hornetq.core.persistence.impl.journal.JournalStorageManager;
import org.hornetq.core.remoting.Channel;
import org.hornetq.core.remoting.Packet;
@@ -25,10 +33,14 @@
import org.hornetq.core.remoting.impl.wireformat.ReplicationCommitMessage;
import org.hornetq.core.remoting.impl.wireformat.ReplicationDeleteMessage;
import org.hornetq.core.remoting.impl.wireformat.ReplicationDeleteTXMessage;
+import org.hornetq.core.remoting.impl.wireformat.ReplicationPageEventMessage;
+import org.hornetq.core.remoting.impl.wireformat.ReplicationPageWriteMessage;
import org.hornetq.core.remoting.impl.wireformat.ReplicationPrepareMessage;
import org.hornetq.core.remoting.impl.wireformat.ReplicationResponseMessage;
import org.hornetq.core.replication.ReplicationEndpoint;
import org.hornetq.core.server.HornetQServer;
+import org.hornetq.core.server.ServerMessage;
+import org.hornetq.utils.SimpleString;
/**
*
@@ -57,6 +69,10 @@
private JournalStorageManager storage;
+ private PagingManager pageManager;
+
+ private final ConcurrentMap<SimpleString, ConcurrentMap<Integer, Page>> pageIndex = new ConcurrentHashMap<SimpleString, ConcurrentMap<Integer, Page>>();
+
// Constructors --------------------------------------------------
public ReplicationEndpointImpl(final HornetQServer server)
{
@@ -74,28 +90,37 @@
{
if (packet.getType() == PacketImpl.REPLICATION_APPEND)
{
- handleAppendAddRecord(packet);
+ handleAppendAddRecord((ReplicationAddMessage)packet);
}
else if (packet.getType() == PacketImpl.REPLICATION_APPEND_TX)
{
- handleAppendAddTXRecord(packet);
+ handleAppendAddTXRecord((ReplicationAddTXMessage)packet);
}
else if (packet.getType() == PacketImpl.REPLICATION_DELETE)
{
- handleAppendDelete(packet);
+ handleAppendDelete((ReplicationDeleteMessage)packet);
}
else if (packet.getType() == PacketImpl.REPLICATION_DELETE_TX)
{
- handleAppendDeleteTX(packet);
+ handleAppendDeleteTX((ReplicationDeleteTXMessage)packet);
}
else if (packet.getType() == PacketImpl.REPLICATION_PREPARE)
{
- handlePrepare(packet);
+ handlePrepare((ReplicationPrepareMessage)packet);
}
else if (packet.getType() == PacketImpl.REPLICATION_COMMIT_ROLLBACK)
{
- handleCommitRollback(packet);
+ handleCommitRollback((ReplicationCommitMessage)packet);
}
+ else if (packet.getType() == PacketImpl.REPLICATION_PAGE_WRITE)
+ {
+ handlePageWrite((ReplicationPageWriteMessage)packet);
+ }
+ else if (packet.getType() == PacketImpl.REPLICATION_PAGE_EVENT)
+ {
+ handlePageEvent((ReplicationPageEventMessage)packet);
+ }
+
}
catch (Exception e)
{
@@ -120,8 +145,7 @@
{
Configuration config = server.getConfiguration();
- // TODO: this needs an executor
- storage = new JournalStorageManager(config, null);
+ storage = new JournalStorageManager(config, server.getExecutorFactory().getExecutor());
storage.start();
bindingsJournal = storage.getBindingsJournal();
@@ -129,6 +153,15 @@
// We only need to load internal structures on the backup...
storage.loadInternalOnly();
+
+ pageManager = new PagingManagerImpl(new PagingStoreFactoryNIO(config.getPagingDirectory(),
+ server.getExecutorFactory()),
+ storage,
+ server.getAddressSettingsRepository(),
+ false);
+
+ pageManager.start();
+
}
/* (non-Javadoc)
@@ -165,82 +198,70 @@
/**
* @param packet
*/
- private void handleCommitRollback(final Packet packet) throws Exception
+ private void handleCommitRollback(final ReplicationCommitMessage packet) throws Exception
{
- ReplicationCommitMessage commitMessage = (ReplicationCommitMessage)packet;
+ Journal journalToUse = getJournal(packet.getJournalID());
- Journal journalToUse = getJournal(commitMessage.getJournalID());
-
- if (commitMessage.isRollback())
+ if (packet.isRollback())
{
- journalToUse.appendRollbackRecord(commitMessage.getTxId(), false);
+ journalToUse.appendRollbackRecord(packet.getTxId(), false);
}
else
{
- journalToUse.appendCommitRecord(commitMessage.getTxId(), false);
+ journalToUse.appendCommitRecord(packet.getTxId(), false);
}
}
/**
* @param packet
*/
- private void handlePrepare(final Packet packet) throws Exception
+ private void handlePrepare(final ReplicationPrepareMessage packet) throws Exception
{
- ReplicationPrepareMessage prepareMessage = (ReplicationPrepareMessage)packet;
+ Journal journalToUse = getJournal(packet.getJournalID());
- Journal journalToUse = getJournal(prepareMessage.getJournalID());
-
- journalToUse.appendPrepareRecord(prepareMessage.getTxId(), prepareMessage.getRecordData(), false);
+ journalToUse.appendPrepareRecord(packet.getTxId(), packet.getRecordData(), false);
}
/**
* @param packet
*/
- private void handleAppendDeleteTX(final Packet packet) throws Exception
+ private void handleAppendDeleteTX(final ReplicationDeleteTXMessage packet) throws Exception
{
- ReplicationDeleteTXMessage deleteMessage = (ReplicationDeleteTXMessage)packet;
+ Journal journalToUse = getJournal(packet.getJournalID());
- Journal journalToUse = getJournal(deleteMessage.getJournalID());
-
- journalToUse.appendDeleteRecordTransactional(deleteMessage.getTxId(),
- deleteMessage.getId(),
- deleteMessage.getRecordData());
+ journalToUse.appendDeleteRecordTransactional(packet.getTxId(), packet.getId(), packet.getRecordData());
}
/**
* @param packet
*/
- private void handleAppendDelete(final Packet packet) throws Exception
+ private void handleAppendDelete(final ReplicationDeleteMessage packet) throws Exception
{
- ReplicationDeleteMessage deleteMessage = (ReplicationDeleteMessage)packet;
+ Journal journalToUse = getJournal(packet.getJournalID());
- Journal journalToUse = getJournal(deleteMessage.getJournalID());
-
- journalToUse.appendDeleteRecord(deleteMessage.getId(), false);
+ journalToUse.appendDeleteRecord(packet.getId(), false);
}
/**
* @param packet
*/
- private void handleAppendAddTXRecord(final Packet packet) throws Exception
+ private void handleAppendAddTXRecord(final ReplicationAddTXMessage packet) throws Exception
{
- ReplicationAddTXMessage addMessage = (ReplicationAddTXMessage)packet;
+ Journal journalToUse = getJournal(packet.getJournalID());
- Journal journalToUse = getJournal(addMessage.getJournalID());
-
- if (addMessage.isUpdate())
+ if (packet.isUpdate())
{
- journalToUse.appendUpdateRecordTransactional(addMessage.getTxId(),
- addMessage.getId(),
- addMessage.getRecordType(),
- addMessage.getRecordData());
+ journalToUse.appendUpdateRecordTransactional(packet.getTxId(),
+ packet.getId(),
+ packet.getRecordType(),
+ packet.getRecordData());
}
else
{
- journalToUse.appendAddRecordTransactional(addMessage.getTxId(),
- addMessage.getId(),
- addMessage.getRecordType(),
- addMessage.getRecordData());
+ journalToUse.appendAddRecordTransactional(packet.getTxId(),
+ packet.getId(),
+ packet.getRecordType(),
+ packet.getRecordData());
}
}
@@ -248,34 +269,109 @@
* @param packet
* @throws Exception
*/
- private void handleAppendAddRecord(final Packet packet) throws Exception
+ private void handleAppendAddRecord(final ReplicationAddMessage packet) throws Exception
{
- ReplicationAddMessage addMessage = (ReplicationAddMessage)packet;
+ Journal journalToUse = getJournal(packet.getJournalID());
- Journal journalToUse = getJournal(addMessage.getJournalID());
-
- if (addMessage.isUpdate())
+ if (packet.isUpdate())
{
if (trace)
{
- System.out.println("Endpoint appendUpdate id = " + addMessage.getId());
+ System.out.println("Endpoint appendUpdate id = " + packet.getId());
}
- journalToUse.appendUpdateRecord(addMessage.getId(),
- addMessage.getRecordType(),
- addMessage.getRecordData(),
- false);
+ journalToUse.appendUpdateRecord(packet.getId(), packet.getRecordType(), packet.getRecordData(), false);
}
else
{
if (trace)
{
- System.out.println("Endpoint append id = " + addMessage.getId());
+ System.out.println("Endpoint append id = " + packet.getId());
}
- journalToUse.appendAddRecord(addMessage.getId(), addMessage.getRecordType(), addMessage.getRecordData(), false);
+ journalToUse.appendAddRecord(packet.getId(), packet.getRecordType(), packet.getRecordData(), false);
}
}
/**
+ * @param packet
+ */
+ private void handlePageEvent(final ReplicationPageEventMessage packet) throws Exception
+ {
+ ConcurrentMap<Integer, Page> pages = getPageMap(packet.getStoreName());
+
+ Page page = pages.remove(packet.getPageNumber());
+
+ if (page != null)
+ {
+ if (packet.isDelete())
+ {
+ page.delete();
+ }
+ else
+ {
+ page.close();
+ }
+ }
+
+ }
+
+ /**
+ * @param packet
+ */
+ private void handlePageWrite(final ReplicationPageWriteMessage packet) throws Exception
+ {
+ PagedMessage pgdMessage = packet.getPagedMessage();
+ ServerMessage msg = pgdMessage.getMessage(storage);
+ Page page = getPage(msg.getDestination(), packet.getPageNumber());
+ page.write(pgdMessage);
+ }
+
+ private ConcurrentMap<Integer, Page> getPageMap(final SimpleString storeName)
+ {
+ ConcurrentMap<Integer, Page> resultIndex = pageIndex.get(storeName);
+
+ if (resultIndex == null)
+ {
+ resultIndex = pageIndex.putIfAbsent(storeName, new ConcurrentHashMap<Integer, Page>());
+ }
+
+ return resultIndex;
+ }
+
+ private Page getPage(final SimpleString storeName, final int pageId) throws Exception
+ {
+ ConcurrentMap<Integer, Page> map = getPageMap(storeName);
+
+ Page page = map.get(pageId);
+
+ if (page == null)
+ {
+ page = newPage(pageId, storeName, map);
+ }
+
+ return page;
+ }
+
+ /**
+ * @param pageId
+ * @param map
+ * @return
+ */
+ private synchronized Page newPage(final int pageId,
+ final SimpleString storeName,
+ final ConcurrentMap<Integer, Page> map) throws Exception
+ {
+ Page page = map.get(pageId);
+
+ if (page == null)
+ {
+ page = pageManager.getPageStore(storeName).createPage(pageId);
+ map.put(pageId, page);
+ }
+
+ return page;
+ }
+
+ /**
* @param journalID
* @return
*/
Modified: branches/Replication_Clebert/src/main/org/hornetq/core/replication/impl/ReplicationManagerImpl.java
===================================================================
--- branches/Replication_Clebert/src/main/org/hornetq/core/replication/impl/ReplicationManagerImpl.java 2009-10-08 17:09:12 UTC (rev 8069)
+++ branches/Replication_Clebert/src/main/org/hornetq/core/replication/impl/ReplicationManagerImpl.java 2009-10-08 21:40:37 UTC (rev 8070)
@@ -21,6 +21,7 @@
import org.hornetq.core.client.impl.ConnectionManager;
import org.hornetq.core.journal.EncodingSupport;
import org.hornetq.core.logging.Logger;
+import org.hornetq.core.paging.PagedMessage;
import org.hornetq.core.remoting.Channel;
import org.hornetq.core.remoting.ChannelHandler;
import org.hornetq.core.remoting.Packet;
@@ -32,11 +33,14 @@
import org.hornetq.core.remoting.impl.wireformat.ReplicationCommitMessage;
import org.hornetq.core.remoting.impl.wireformat.ReplicationDeleteMessage;
import org.hornetq.core.remoting.impl.wireformat.ReplicationDeleteTXMessage;
+import org.hornetq.core.remoting.impl.wireformat.ReplicationPageEventMessage;
+import org.hornetq.core.remoting.impl.wireformat.ReplicationPageWriteMessage;
import org.hornetq.core.remoting.impl.wireformat.ReplicationPrepareMessage;
import org.hornetq.core.remoting.spi.HornetQBuffer;
import org.hornetq.core.replication.ReplicationManager;
import org.hornetq.core.replication.ReplicationToken;
import org.hornetq.utils.ConcurrentHashSet;
+import org.hornetq.utils.SimpleString;
/**
* A RepplicationManagerImpl
@@ -217,6 +221,39 @@
}
/* (non-Javadoc)
+ * @see org.hornetq.core.replication.ReplicationManager#pageClosed(org.hornetq.utils.SimpleString, int)
+ */
+ public void pageClosed(final SimpleString storeName, final int pageNumber)
+ {
+ if (enabled)
+ {
+ sendReplicatePacket(new ReplicationPageEventMessage(storeName, pageNumber, false));
+ }
+ }
+
+ /* (non-Javadoc)
+ * @see org.hornetq.core.replication.ReplicationManager#pageDeleted(org.hornetq.utils.SimpleString, int)
+ */
+ public void pageDeleted(final SimpleString storeName, final int pageNumber)
+ {
+ if (enabled)
+ {
+ sendReplicatePacket(new ReplicationPageEventMessage(storeName, pageNumber, true));
+ }
+ }
+
+ /* (non-Javadoc)
+ * @see org.hornetq.core.replication.ReplicationManager#pageWrite(org.hornetq.utils.SimpleString, int)
+ */
+ public void pageWrite(final PagedMessage message, final int pageNumber)
+ {
+ if (enabled)
+ {
+ sendReplicatePacket(new ReplicationPageWriteMessage(message, pageNumber));
+ }
+ }
+
+ /* (non-Javadoc)
* @see org.hornetq.core.server.HornetQComponent#isStarted()
*/
public synchronized boolean isStarted()
Modified: branches/Replication_Clebert/tests/src/org/hornetq/tests/integration/paging/PageCrashTest.java
===================================================================
--- branches/Replication_Clebert/tests/src/org/hornetq/tests/integration/paging/PageCrashTest.java 2009-10-08 17:09:12 UTC (rev 8069)
+++ branches/Replication_Clebert/tests/src/org/hornetq/tests/integration/paging/PageCrashTest.java 2009-10-08 21:40:37 UTC (rev 8070)
@@ -310,7 +310,7 @@
}
@Override
- protected Page createPage(final int page) throws Exception
+ public Page createPage(final int page) throws Exception
{
Page originalPage = super.createPage(page);
More information about the hornetq-commits
mailing list