Author: borges Date: 2011-10-10 09:35:11 -0400 (Mon, 10 Oct 2011) New Revision: 11501 Modified: branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java Log: HORNETQ-720 Make ReplicationEndpoint.stop() more robust, and do close all open files Modified: branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java =================================================================== --- branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java 2011-10-10 13:35:00 UTC (rev 11500) +++ branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java 2011-10-10 13:35:11 UTC (rev 11501) @@ -18,6 +18,7 @@ import java.io.IOException; import java.nio.ByteBuffer; import java.nio.channels.FileChannel; +import java.util.Arrays; import java.util.EnumSet; import java.util.HashMap; import java.util.Map; @@ -91,7 +92,7 @@ /** Files reserved in each journal for synchronization of existing data from the 'live' server. */ private final Map<JournalContent, Map<Long, JournalSyncFile>> filesReservedForSync = new HashMap<JournalContent, Map<Long, JournalSyncFile>>(); - private Map<Long, LargeServerMessage> largeMessagesOnSync = new HashMap<Long, LargeServerMessage>(); + private final Map<Long, LargeServerMessage> largeMessagesOnSync = new HashMap<Long, LargeServerMessage>(); /** * Used to hold the real Journals before the backup is synchronized. This field should be @@ -144,6 +145,7 @@ */ public void handlePacket(final Packet packet) { + PacketImpl response = new ReplicationResponseMessage(); final byte type=packet.getType(); @@ -230,7 +232,7 @@ return started; } - public void start() throws Exception + public synchronized void start() throws Exception { Configuration config = server.getConfiguration(); @@ -263,22 +265,19 @@ } - /* (non-Javadoc) - * @see org.hornetq.core.server.HornetQComponent#stop() - */ - public void stop() throws Exception + public synchronized void stop() throws Exception { - if(!started) + if (!started) { return; } + // This could be null if the backup server is being // shut down without any live server connecting here if (channel != null) { channel.close(); } - storage.stop(); for (ConcurrentMap<Integer, Page> map : pageIndex.values()) { @@ -301,11 +300,28 @@ { largeMessage.releaseResources(); } - largeMessages.clear(); + for (LargeServerMessage largeMessage : largeMessagesOnSync.values()) + { + largeMessage.releaseResources(); + } + largeMessagesOnSync.clear(); + + for (Entry<JournalContent, Map<Long, JournalSyncFile>> entry : filesReservedForSync.entrySet()) + { + for (JournalSyncFile filesReserved : entry.getValue().values()) + { + filesReserved.close(); + } + } + + filesReservedForSync.clear(); pageManager.stop(); + // Storage needs to be the last to stop + storage.stop(); + started = false; } @@ -437,9 +453,9 @@ } largeMessages.clear(); largeMessages.putAll(largeMessagesOnSync); + largeMessagesOnSync.clear(); } } - largeMessagesOnSync = null; journalsHolder = null; server.setRemoteBackupUpToDate(); log.info("Backup server " + server + " is synchronized with live-server."); @@ -524,15 +540,23 @@ return; } + final Journal journal = journalsHolder.get(packet.getJournalContentType()); + synchronized (this) + { + if (!started) + return; + Map<Long, JournalSyncFile> mapToFill = filesReservedForSync.get(packet.getJournalContentType()); + log.info("Journal " + packet.getJournalContentType() + ". Reserving fileIDs for synchronization: " + + Arrays.toString(packet.getFileIds())); - Map<Long, JournalSyncFile> mapToFill = filesReservedForSync.get(packet.getJournalContentType()); for (Entry<Long, JournalFile> entry : journal.createFilesForBackupSync(packet.getFileIds()).entrySet()) { mapToFill.put(entry.getKey(), new JournalSyncFile(entry.getValue())); } registerJournal(packet.getJournalContentType().typeByte, new FileWrapperJournal(journal)); - } + } + } private void handleLargeMessageEnd(final ReplicationLargeMessageEndMessage packet) { @@ -601,9 +625,7 @@ log.warn("Large MessageID " + messageId + " is not available on backup server. Ignoring replication message"); } - return message; - } /** @@ -845,7 +867,8 @@ void close() throws IOException { - channel.close(); + if (channel != null) + channel.close(); } @Override