anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Minor change

Browse Source
This commit is contained in:
andsel
2015-09-29 21:39:32 +02:00
parent 3e37df1275
commit db6d296c20
2 changed files with 26 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java
+4 -6
View File
@@ -124,7 +124,7 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
public Collection<StoredMessage> searchMatching(IMatchingCondition condition) {
LOG.debug("searchMatching scanning all retained messages, presents are {}", m_retainedStore.size());
List<StoredMessage> results = new ArrayList<StoredMessage>();
List<StoredMessage> results = new ArrayList<>();
for (Map.Entry<String, StoredMessage> entry : m_retainedStore.entrySet()) {
StoredMessage storedMsg = entry.getValue();
@@ -174,7 +174,7 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
//was a qos0 message (no ID)
toRemoveEvt = evt;
}
if (evt.getMessageID() == messageID) {
if (evt.getMessageID().equals(messageID)) {
toRemoveEvt = evt;
}
}
@@ -333,16 +333,14 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
}
private StoredPublishEvent convertToStored(PublishEvent evt) {
StoredPublishEvent storedEvt = new StoredPublishEvent(evt);
return storedEvt;
return new StoredPublishEvent(evt);
}
private PublishEvent convertFromStored(StoredPublishEvent evt) {
byte[] message = evt.getMessage();
ByteBuffer bbmessage = ByteBuffer.wrap(message);
//bbmessage.flip();
PublishEvent liveEvt = new PublishEvent(evt.getTopic(), evt.getQos(),
return new PublishEvent(evt.getTopic(), evt.getQos(),
bbmessage, evt.isRetain(), evt.getClientID(), evt.getMessageID());
return liveEvt;
}
}
broker/src/test/java/org/eclipse/moquette/server/ServerShutdownTest.java
+22 -8
View File
@@ -1,3 +1,18 @@
/*
* Copyright (c) 2012-2015 The original author or authors
* ------------------------------------------------------
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* and Apache License v2.0 which accompanies this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v10.html
*
* The Apache License v2.0 is available at
* http://www.opensource.org/licenses/apache2.0.php
*
* You may elect to redistribute this code under either of these licenses.
*/
package org.eclipse.moquette.server;
import org.junit.Test;
@@ -24,10 +39,9 @@ public class ServerShutdownTest {
@Test
public void testShutdown() throws IOException, InterruptedException {
final int initialThreadCount = Thread.activeCount();
LOG.info("*** testShutdown ***");
LOG.debug("Initial Thread Count = " + initialThreadCount);
LOG.debug("Initial threads count {}", initialThreadCount);
//Start the server
Server broker = new Server();
@@ -40,12 +54,12 @@ public class ServerShutdownTest {
threadCounter.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
int threadCount = Thread.activeCount();
LOG.debug("Current Thread Count = " + threadCount);
//plus 1 for this thread
if (threadCount == initialThreadCount + 1) {
threadsStoppedLatch.countDown();
}
int threadCount = Thread.activeCount();
LOG.debug("Current Thread Count = " + threadCount);
//plus 1 for this thread
if (threadCount == initialThreadCount + 1) {
threadsStoppedLatch.countDown();
}
}
}, 0, 100, TimeUnit.MILLISECONDS);