You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by cl...@apache.org on 2015/08/10 17:13:02 UTC
[03/53] [abbrv] [partial] activemq-artemis git commit: automatic
checkstyle change
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/bac96047/artemis-native/src/main/java/org/apache/activemq/artemis/jlibaio/LibaioContext.java
----------------------------------------------------------------------
diff --git a/artemis-native/src/main/java/org/apache/activemq/artemis/jlibaio/LibaioContext.java b/artemis-native/src/main/java/org/apache/activemq/artemis/jlibaio/LibaioContext.java
index 8b45f54..5ada655 100644
--- a/artemis-native/src/main/java/org/apache/activemq/artemis/jlibaio/LibaioContext.java
+++ b/artemis-native/src/main/java/org/apache/activemq/artemis/jlibaio/LibaioContext.java
@@ -40,8 +40,7 @@ import static io.netty.util.internal.ObjectUtil.checkNotNull;
* <br>
* <a href="https://ext4.wiki.kernel.org/index.php/Clarifying_Direct_IO's_Semantics">Interesting reading for this.</a>
*/
-public class LibaioContext <Callback extends SubmitInfo> implements Closeable
-{
+public class LibaioContext<Callback extends SubmitInfo> implements Closeable {
private static final AtomicLong totalMaxIO = new AtomicLong(0);
@@ -54,80 +53,69 @@ public class LibaioContext <Callback extends SubmitInfo> implements Closeable
private static boolean loaded = false;
- public static boolean isLoaded()
- {
+ public static boolean isLoaded() {
return loaded;
}
- private static boolean loadLibrary(final String name)
- {
- try
- {
+ private static boolean loadLibrary(final String name) {
+ try {
System.loadLibrary(name);
- if (getNativeVersion() != EXPECTED_NATIVE_VERSION)
- {
+ if (getNativeVersion() != EXPECTED_NATIVE_VERSION) {
NativeLogger.LOGGER.incompatibleNativeLibrary();
return false;
}
- else
- {
+ else {
return true;
}
}
- catch (Throwable e)
- {
+ catch (Throwable e) {
NativeLogger.LOGGER.debug(name + " -> error loading the native library", e);
return false;
}
}
- static
- {
+ static {
String[] libraries = new String[]{"artemis-native-64", "artemis-native-32"};
- for (String library : libraries)
- {
- if (loadLibrary(library))
- {
+ for (String library : libraries) {
+ if (loadLibrary(library)) {
loaded = true;
break;
}
- else
- {
+ else {
NativeLogger.LOGGER.debug("Library " + library + " not found!");
}
}
- if (!loaded)
- {
+ if (!loaded) {
NativeLogger.LOGGER.debug("Couldn't locate LibAIO Wrapper");
}
}
/**
* This is used to validate leaks on tests.
+ *
* @return the number of allocated aio, to be used on test checks.
*/
- public static long getTotalMaxIO()
- {
+ public static long getTotalMaxIO() {
return totalMaxIO.get();
}
/**
* It will reset all the positions on the buffer to 0, using memset.
+ *
* @param buffer a native buffer.
-s */
- public void memsetBuffer(ByteBuffer buffer)
- {
+ * s
+ */
+ public void memsetBuffer(ByteBuffer buffer) {
memsetBuffer(buffer, buffer.limit());
}
/**
* This is used on tests validating for leaks.
*/
- public static void resetMaxAIO()
- {
+ public static void resetMaxAIO() {
totalMaxIO.set(0);
}
@@ -146,79 +134,71 @@ s */
* The queue size here will use resources defined on the kernel parameter
* <a href="https://www.kernel.org/doc/Documentation/sysctl/fs.txt">fs.aio-max-nr</a> .
*
- * @param queueSize the size to be initialize on libaio
- * io_queue_init which can't be higher than /proc/sys/fs/aio-max-nr.
+ * @param queueSize the size to be initialize on libaio
+ * io_queue_init which can't be higher than /proc/sys/fs/aio-max-nr.
* @param useSemaphore should block on a semaphore avoiding using more submits than what's available.
*/
- public LibaioContext(int queueSize, boolean useSemaphore)
- {
- try
- {
+ public LibaioContext(int queueSize, boolean useSemaphore) {
+ try {
this.ioContext = newContext(queueSize);
}
- catch (Exception e)
- {
+ catch (Exception e) {
throw e;
}
this.queueSize = queueSize;
totalMaxIO.addAndGet(queueSize);
- if (useSemaphore)
- {
+ if (useSemaphore) {
this.ioSpace = new Semaphore(queueSize);
}
- else
- {
+ else {
this.ioSpace = null;
}
}
-
/**
* Documented at {@link LibaioFile#write(long, int, java.nio.ByteBuffer, SubmitInfo)}
- * @param fd the file descriptor
- * @param position the write position
- * @param size number of bytes to use
+ *
+ * @param fd the file descriptor
+ * @param position the write position
+ * @param size number of bytes to use
* @param bufferWrite the native buffer
- * @param callback a callback
+ * @param callback a callback
* @throws IOException in case of error
*/
- public void submitWrite(int fd,long position, int size,
- ByteBuffer bufferWrite, Callback callback) throws IOException
- {
- try
- {
- if (ioSpace != null)
- {
+ public void submitWrite(int fd,
+ long position,
+ int size,
+ ByteBuffer bufferWrite,
+ Callback callback) throws IOException {
+ try {
+ if (ioSpace != null) {
ioSpace.acquire();
}
}
- catch (InterruptedException e)
- {
+ catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new IOException(e.getMessage(), e);
}
submitWrite(fd, this.ioContext, position, size, bufferWrite, callback);
}
- public void submitRead(int fd, long position, int size, ByteBuffer bufferWrite,
- Callback callback) throws IOException
- {
- try
- {
- if (ioSpace != null)
- {
+ public void submitRead(int fd,
+ long position,
+ int size,
+ ByteBuffer bufferWrite,
+ Callback callback) throws IOException {
+ try {
+ if (ioSpace != null) {
ioSpace.acquire();
}
}
- catch (InterruptedException e)
- {
+ catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new IOException(e.getMessage(), e);
}
submitRead(fd, this.ioContext, position, size, bufferWrite, callback);
}
-
/**
* This is used to close the libaio queues and cleanup the native data used.
* <br>
@@ -226,22 +206,18 @@ s */
* this could cause core dumps or VM crashes.
*/
@Override
- public void close()
- {
- if (!closed.getAndSet(true))
- {
+ public void close() {
+ if (!closed.getAndSet(true)) {
totalMaxIO.addAndGet(-queueSize);
- if (ioContext != null)
- {
+ if (ioContext != null) {
deleteContext(ioContext);
}
}
}
@Override
- protected void finalize() throws Throwable
- {
+ protected void finalize() throws Throwable {
super.finalize();
close();
}
@@ -250,13 +226,12 @@ s */
* It will open a file. If you set the direct flag = false then you won't need to use the special buffer.
* Notice: This will create an empty file if the file doesn't already exist.
*
- * @param file the file to be open.
+ * @param file the file to be open.
* @param direct will set ODIRECT.
* @return It will return a LibaioFile instance.
* @throws IOException in case of error.
*/
- public LibaioFile<Callback> openFile(File file, boolean direct) throws IOException
- {
+ public LibaioFile<Callback> openFile(File file, boolean direct) throws IOException {
return openFile(file.getPath(), direct);
}
@@ -264,13 +239,12 @@ s */
* It will open a file. If you set the direct flag = false then you won't need to use the special buffer.
* Notice: This will create an empty file if the file doesn't already exist.
*
- * @param file the file to be open.
+ * @param file the file to be open.
* @param direct should use O_DIRECT when opening the file.
* @return a new open file.
* @throws IOException in case of error.
*/
- public LibaioFile<Callback> openFile(String file, boolean direct) throws IOException
- {
+ public LibaioFile<Callback> openFile(String file, boolean direct) throws IOException {
checkNotNull(file, "path");
checkNotNull(ioContext, "IOContext");
@@ -283,13 +257,13 @@ s */
/**
* It will open a file disassociated with any sort of factory.
* This is useful when you won't use reading / writing through libaio like locking files.
- * @param file a file name
+ *
+ * @param file a file name
* @param direct will use O_DIRECT
* @return a new file
* @throws IOException in case of error.
*/
- public static LibaioFile openControlFile(String file, boolean direct) throws IOException
- {
+ public static LibaioFile openControlFile(String file, boolean direct) throws IOException {
checkNotNull(file, "path");
// note: the native layer will throw an IOException in case of errors
@@ -306,7 +280,7 @@ s */
* Thread polling for any reason.
* <br>
* Notice that the native layer will invoke {@link SubmitInfo#onError(int, String)} in case of failures,
- * but it won't call done method for you.
+ * but it won't call done method for you.
*
* @param callbacks area to receive the callbacks passed on submission.The size of this callback has to
* be greater than the parameter max.
@@ -316,36 +290,32 @@ s */
* @see LibaioFile#write(long, int, java.nio.ByteBuffer, SubmitInfo)
* @see LibaioFile#read(long, int, java.nio.ByteBuffer, SubmitInfo)
*/
- public int poll(Callback[] callbacks, int min, int max)
- {
+ public int poll(Callback[] callbacks, int min, int max) {
int released = poll(ioContext, callbacks, min, max);
- if (ioSpace != null)
- {
- if (released > 0)
- {
+ if (ioSpace != null) {
+ if (released > 0) {
ioSpace.release(released);
}
}
- return released;
+ return released;
}
/**
* It will start polling and will keep doing until the context is closed.
* This will call callbacks on {@link SubmitInfo#onError(int, String)} and
- * {@link SubmitInfo#done()}.
+ * {@link SubmitInfo#done()}.
* In case of error, both {@link SubmitInfo#onError(int, String)} and
- * {@link SubmitInfo#done()} are called.
+ * {@link SubmitInfo#done()} are called.
*/
- public void poll()
- {
+ public void poll() {
blockedPoll(ioContext);
}
- /** Called from the native layer */
- private void done(SubmitInfo info)
- {
- if (ioSpace != null)
- {
+ /**
+ * Called from the native layer
+ */
+ private void done(SubmitInfo info) {
+ if (ioSpace != null) {
ioSpace.release();
}
info.done();
@@ -364,7 +334,7 @@ s */
/**
* it will return a file descriptor.
*
- * @param path the file name.
+ * @param path the file name.
* @param direct translates as O_DIRECT On open
* @return a fd from open C call.
*/
@@ -380,7 +350,7 @@ s */
* <br>
* Documented at {@link LibaioFile#newBuffer(int)}.
*
- * @param size needs to be % alignment
+ * @param size needs to be % alignment
* @param alignment the alignment used at the dispositive
* @return a new native buffer used with posix_memalign
*/
@@ -388,6 +358,7 @@ s */
/**
* This will call posix free to release the inner buffer allocated at {@link #newAlignedBuffer(int, int)}.
+ *
* @param buffer a native buffer allocated with {@link #newAlignedBuffer(int, int)}.
*/
public static native void freeBuffer(ByteBuffer buffer);
@@ -396,17 +367,21 @@ s */
* Documented at {@link LibaioFile#write(long, int, java.nio.ByteBuffer, SubmitInfo)}.
*/
native void submitWrite(int fd,
- ByteBuffer libaioContext,
- long position, int size, ByteBuffer bufferWrite,
- Callback callback) throws IOException;
+ ByteBuffer libaioContext,
+ long position,
+ int size,
+ ByteBuffer bufferWrite,
+ Callback callback) throws IOException;
/**
* Documented at {@link LibaioFile#read(long, int, java.nio.ByteBuffer, SubmitInfo)}.
*/
native void submitRead(int fd,
- ByteBuffer libaioContext,
- long position, int size, ByteBuffer bufferWrite,
- Callback callback) throws IOException;
+ ByteBuffer libaioContext,
+ long position,
+ int size,
+ ByteBuffer bufferWrite,
+ Callback callback) throws IOException;
/**
* Note: this shouldn't be done concurrently.
@@ -431,8 +406,7 @@ s */
static native int getBlockSizeFD(int fd);
- public static int getBlockSize(File path)
- {
+ public static int getBlockSize(File path) {
return getBlockSize(path.getAbsolutePath());
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/bac96047/artemis-native/src/main/java/org/apache/activemq/artemis/jlibaio/LibaioFile.java
----------------------------------------------------------------------
diff --git a/artemis-native/src/main/java/org/apache/activemq/artemis/jlibaio/LibaioFile.java b/artemis-native/src/main/java/org/apache/activemq/artemis/jlibaio/LibaioFile.java
index bf88d65..27243fe 100644
--- a/artemis-native/src/main/java/org/apache/activemq/artemis/jlibaio/LibaioFile.java
+++ b/artemis-native/src/main/java/org/apache/activemq/artemis/jlibaio/LibaioFile.java
@@ -22,8 +22,8 @@ import java.nio.ByteBuffer;
/**
* This is an extension to use libaio.
*/
-public final class LibaioFile<Callback extends SubmitInfo>
-{
+public final class LibaioFile<Callback extends SubmitInfo> {
+
protected boolean open;
/**
* This represents a structure allocated on the native
@@ -33,29 +33,25 @@ public final class LibaioFile<Callback extends SubmitInfo>
private int fd;
- LibaioFile(int fd, LibaioContext ctx)
- {
+ LibaioFile(int fd, LibaioContext ctx) {
this.ctx = ctx;
this.fd = fd;
}
- public int getBlockSize()
- {
+ public int getBlockSize() {
return 512;
// FIXME
//return LibaioContext.getBlockSizeFD(fd);
}
- public boolean lock()
- {
+ public boolean lock() {
return LibaioContext.lock(fd);
}
/**
* {@inheritDoc}
*/
- public void close() throws IOException
- {
+ public void close() throws IOException {
open = false;
LibaioContext.close(fd);
}
@@ -63,8 +59,7 @@ public final class LibaioFile<Callback extends SubmitInfo>
/**
* @return The size of the file.
*/
- public long getSize()
- {
+ public long getSize() {
return LibaioContext.getSize(fd);
}
@@ -82,8 +77,7 @@ public final class LibaioFile<Callback extends SubmitInfo>
* @param callback A callback to be returned on the poll method.
* @throws java.io.IOException in case of error
*/
- public void write(long position, int size, ByteBuffer buffer, Callback callback) throws IOException
- {
+ public void write(long position, int size, ByteBuffer buffer, Callback callback) throws IOException {
ctx.submitWrite(fd, position, size, buffer, callback);
}
@@ -103,8 +97,7 @@ public final class LibaioFile<Callback extends SubmitInfo>
* @throws java.io.IOException in case of error
* @see LibaioContext#poll(SubmitInfo[], int, int)
*/
- public void read(long position, int size, ByteBuffer buffer, Callback callback) throws IOException
- {
+ public void read(long position, int size, ByteBuffer buffer, Callback callback) throws IOException {
ctx.submitRead(fd, position, size, buffer, callback);
}
@@ -118,23 +111,20 @@ public final class LibaioFile<Callback extends SubmitInfo>
* @param size the size of the buffer.
* @return the buffer allocated.
*/
- public ByteBuffer newBuffer(int size)
- {
+ public ByteBuffer newBuffer(int size) {
return LibaioContext.newAlignedBuffer(size, 512);
}
/**
* It will preallocate the file with a given size.
+ *
* @param size number of bytes to be filled on the file
*/
- public void fill(long size)
- {
- try
- {
+ public void fill(long size) {
+ try {
LibaioContext.fill(fd, size);
}
- catch (OutOfMemoryError e)
- {
+ catch (OutOfMemoryError e) {
NativeLogger.LOGGER.debug("Didn't have enough memory to allocate " + size + " bytes in memory, using simple fallocate");
LibaioContext.fallocate(fd, size);
}
@@ -142,10 +132,10 @@ public final class LibaioFile<Callback extends SubmitInfo>
/**
* It will use fallocate to initialize a file.
+ *
* @param size number of bytes to be filled on the file
*/
- public void fallocate(long size)
- {
+ public void fallocate(long size) {
LibaioContext.fallocate(fd, size);
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/bac96047/artemis-native/src/main/java/org/apache/activemq/artemis/jlibaio/NativeLogger.java
----------------------------------------------------------------------
diff --git a/artemis-native/src/main/java/org/apache/activemq/artemis/jlibaio/NativeLogger.java b/artemis-native/src/main/java/org/apache/activemq/artemis/jlibaio/NativeLogger.java
index 449c168..6ebcc8f 100644
--- a/artemis-native/src/main/java/org/apache/activemq/artemis/jlibaio/NativeLogger.java
+++ b/artemis-native/src/main/java/org/apache/activemq/artemis/jlibaio/NativeLogger.java
@@ -37,14 +37,13 @@ import org.jboss.logging.annotations.MessageLogger;
* so an INFO message would be 1000 to 6000
*/
@MessageLogger(projectCode = "jlibaio")
-public interface NativeLogger extends BasicLogger
-{
+public interface NativeLogger extends BasicLogger {
+
/**
* The journal logger.
*/
NativeLogger LOGGER = Logger.getMessageLogger(NativeLogger.class, NativeLogger.class.getPackage().getName());
-
@LogMessage(level = Logger.Level.WARN)
@Message(id = 1001, value = "You have a native library with a different version than expected", format = Message.Format.MESSAGE_FORMAT)
void incompatibleNativeLibrary();
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/bac96047/artemis-native/src/main/java/org/apache/activemq/artemis/jlibaio/SubmitInfo.java
----------------------------------------------------------------------
diff --git a/artemis-native/src/main/java/org/apache/activemq/artemis/jlibaio/SubmitInfo.java b/artemis-native/src/main/java/org/apache/activemq/artemis/jlibaio/SubmitInfo.java
index 47feea8..36d2103 100644
--- a/artemis-native/src/main/java/org/apache/activemq/artemis/jlibaio/SubmitInfo.java
+++ b/artemis-native/src/main/java/org/apache/activemq/artemis/jlibaio/SubmitInfo.java
@@ -17,8 +17,8 @@
package org.apache.activemq.artemis.jlibaio;
-public interface SubmitInfo
-{
+public interface SubmitInfo {
+
void onError(int errno, String message);
void done();
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/bac96047/artemis-native/src/main/java/org/apache/activemq/artemis/jlibaio/util/CallbackCache.java
----------------------------------------------------------------------
diff --git a/artemis-native/src/main/java/org/apache/activemq/artemis/jlibaio/util/CallbackCache.java b/artemis-native/src/main/java/org/apache/activemq/artemis/jlibaio/util/CallbackCache.java
index ec2a630..00a3826 100644
--- a/artemis-native/src/main/java/org/apache/activemq/artemis/jlibaio/util/CallbackCache.java
+++ b/artemis-native/src/main/java/org/apache/activemq/artemis/jlibaio/util/CallbackCache.java
@@ -17,14 +17,13 @@
package org.apache.activemq.artemis.jlibaio.util;
-
import org.apache.activemq.artemis.jlibaio.SubmitInfo;
/**
* this is an utility class where you can reuse Callbackk objects for your LibaioContext usage.
*/
-public class CallbackCache<Callback extends SubmitInfo>
-{
+public class CallbackCache<Callback extends SubmitInfo> {
+
private final SubmitInfo[] pool;
private int put = 0;
@@ -34,34 +33,26 @@ public class CallbackCache<Callback extends SubmitInfo>
private final Object lock = new Object();
- public CallbackCache(int size)
- {
+ public CallbackCache(int size) {
this.pool = new SubmitInfo[size];
this.size = size;
}
- public Callback get()
- {
- synchronized (lock)
- {
- if (available <= 0)
- {
+ public Callback get() {
+ synchronized (lock) {
+ if (available <= 0) {
return null;
}
- else
- {
- Callback retValue = (Callback)pool[get];
+ else {
+ Callback retValue = (Callback) pool[get];
pool[get] = null;
- if (retValue == null)
- {
+ if (retValue == null) {
throw new NullPointerException("You should initialize the pool before using it");
}
- if (retValue != null)
- {
+ if (retValue != null) {
available--;
get++;
- if (get >= size)
- {
+ if (get >= size) {
get = 0;
}
}
@@ -70,20 +61,15 @@ public class CallbackCache<Callback extends SubmitInfo>
}
}
- public CallbackCache put(Callback callback)
- {
- if (callback == null)
- {
+ public CallbackCache put(Callback callback) {
+ if (callback == null) {
return null;
}
- synchronized (lock)
- {
- if (available < size)
- {
+ synchronized (lock) {
+ if (available < size) {
available++;
pool[put++] = callback;
- if (put >= size)
- {
+ if (put >= size) {
put = 0;
}
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/bac96047/artemis-native/src/test/java/org/apache/activemq/artemis/jlibaio/test/CallbackCachelTest.java
----------------------------------------------------------------------
diff --git a/artemis-native/src/test/java/org/apache/activemq/artemis/jlibaio/test/CallbackCachelTest.java b/artemis-native/src/test/java/org/apache/activemq/artemis/jlibaio/test/CallbackCachelTest.java
index f62f8e2..6785d78 100644
--- a/artemis-native/src/test/java/org/apache/activemq/artemis/jlibaio/test/CallbackCachelTest.java
+++ b/artemis-native/src/test/java/org/apache/activemq/artemis/jlibaio/test/CallbackCachelTest.java
@@ -24,16 +24,13 @@ import org.apache.activemq.artemis.jlibaio.SubmitInfo;
import org.junit.Assert;
import org.junit.Test;
-public class CallbackCachelTest
-{
+public class CallbackCachelTest {
+
@Test
- public void testPartiallyInitialized()
- {
+ public void testPartiallyInitialized() {
CallbackCache<MyPool> pool = new CallbackCache(100);
-
- for (int i = 0; i < 50; i++)
- {
+ for (int i = 0; i < 50; i++) {
pool.put(new MyPool(i));
}
@@ -43,69 +40,59 @@ public class CallbackCachelTest
pool.put(value);
-
// add and remove immediately
- for (int i = 0; i < 777; i++)
- {
+ for (int i = 0; i < 777; i++) {
pool.put(pool.get());
}
-
HashSet<MyPool> hashValues = new HashSet<>();
-
MyPool getValue;
- while ((getValue = pool.get()) != null)
- {
+ while ((getValue = pool.get()) != null) {
hashValues.add(getValue);
}
-
Assert.assertEquals(50, hashValues.size());
}
- static class MyPool implements SubmitInfo
- {
+ static class MyPool implements SubmitInfo {
+
public final int i;
- MyPool(int i)
- {
+ MyPool(int i) {
this.i = i;
}
-
- public int getI()
- {
+ public int getI() {
return i;
}
@Override
- public void onError(int errno, String message)
- {
+ public void onError(int errno, String message) {
}
@Override
- public void done()
- {
+ public void done() {
}
@Override
- public boolean equals(Object o)
- {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
+ public boolean equals(Object o) {
+ if (this == o)
+ return true;
+ if (o == null || getClass() != o.getClass())
+ return false;
MyPool myPool = (MyPool) o;
- if (i != myPool.i) return false;
+ if (i != myPool.i)
+ return false;
return true;
}
@Override
- public int hashCode()
- {
+ public int hashCode() {
return i;
}
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/bac96047/artemis-native/src/test/java/org/apache/activemq/artemis/jlibaio/test/LibaioTest.java
----------------------------------------------------------------------
diff --git a/artemis-native/src/test/java/org/apache/activemq/artemis/jlibaio/test/LibaioTest.java b/artemis-native/src/test/java/org/apache/activemq/artemis/jlibaio/test/LibaioTest.java
index e46caf3..c93670a 100644
--- a/artemis-native/src/test/java/org/apache/activemq/artemis/jlibaio/test/LibaioTest.java
+++ b/artemis-native/src/test/java/org/apache/activemq/artemis/jlibaio/test/LibaioTest.java
@@ -41,16 +41,15 @@ import org.junit.rules.TemporaryFolder;
* This test is using a different package from {@link LibaioFile}
* as I need to validate public methods on the API
*/
-public class LibaioTest
-{
+public class LibaioTest {
@BeforeClass
- public static void testAIO()
- {
+ public static void testAIO() {
Assume.assumeTrue(LibaioContext.isLoaded());
}
- /** This is just an arbitrary number for a number of elements you need to pass to the libaio init method
+ /**
+ * This is just an arbitrary number for a number of elements you need to pass to the libaio init method
* Some of the tests are using half of this number, so if anyone decide to change this please use an even number.
*/
private static final int LIBAIO_QUEUE_SIZE = 50;
@@ -61,25 +60,21 @@ public class LibaioTest
public LibaioContext<TestInfo> control;
@Before
- public void setUpFactory()
- {
+ public void setUpFactory() {
control = new LibaioContext<>(LIBAIO_QUEUE_SIZE, true);
}
@After
- public void deleteFactory()
- {
+ public void deleteFactory() {
control.close();
validateLibaio();
}
- public void validateLibaio()
- {
+ public void validateLibaio() {
Assert.assertEquals(0, LibaioContext.getTotalMaxIO());
}
- public LibaioTest()
- {
+ public LibaioTest() {
/*
* I didn't use /tmp for three reasons
* - Most systems now will use tmpfs which is not compatible with O_DIRECT
@@ -92,15 +87,13 @@ public class LibaioTest
}
@Test
- public void testOpen() throws Exception
- {
+ public void testOpen() throws Exception {
LibaioFile fileDescriptor = control.openFile(temporaryFolder.newFile("test.bin"), true);
fileDescriptor.close();
}
@Test
- public void testInitAndFallocate() throws Exception
- {
+ public void testInitAndFallocate() throws Exception {
LibaioFile fileDescriptor = control.openFile(temporaryFolder.newFile("test.bin"), true);
fileDescriptor.fallocate(1024 * 1024);
@@ -112,7 +105,6 @@ public class LibaioTest
fileDescriptor.close();
-
buffer.position(0);
LibaioFile fileDescriptor2 = control.openFile(temporaryFolder.newFile("test2.bin"), true);
@@ -120,8 +112,7 @@ public class LibaioTest
fileDescriptor2.read(0, 1024 * 1024, buffer, new TestInfo());
control.poll(callbacks, 1, 1);
- for (int i = 0; i < 1024 * 1024; i++)
- {
+ for (int i = 0; i < 1024 * 1024; i++) {
Assert.assertEquals(0, buffer.get());
}
@@ -129,8 +120,7 @@ public class LibaioTest
}
@Test
- public void testSubmitWriteOnTwoFiles() throws Exception
- {
+ public void testSubmitWriteOnTwoFiles() throws Exception {
File file1 = temporaryFolder.newFile("test.bin");
File file2 = temporaryFolder.newFile("test2.bin");
@@ -138,8 +128,7 @@ public class LibaioTest
fillupFile(file1, LIBAIO_QUEUE_SIZE / 2);
fillupFile(file2, LIBAIO_QUEUE_SIZE / 2);
- LibaioFile[] fileDescriptor = new LibaioFile[]{control.openFile(file1, true),
- control.openFile(file2, true)};
+ LibaioFile[] fileDescriptor = new LibaioFile[]{control.openFile(file1, true), control.openFile(file2, true)};
Assert.assertEquals((LIBAIO_QUEUE_SIZE / 2) * 512, fileDescriptor[0].getSize());
Assert.assertEquals((LIBAIO_QUEUE_SIZE / 2) * 512, fileDescriptor[1].getSize());
@@ -151,44 +140,36 @@ public class LibaioTest
ByteBuffer buffer = LibaioContext.newAlignedBuffer(512, 512);
- try
- {
- for (int i = 0; i < 512; i++)
- {
+ try {
+ for (int i = 0; i < 512; i++) {
buffer.put((byte) 'a');
}
TestInfo callback = new TestInfo();
TestInfo[] callbacks = new TestInfo[LIBAIO_QUEUE_SIZE];
- for (int i = 0; i < LIBAIO_QUEUE_SIZE / 2; i++)
- {
- for (LibaioFile file : fileDescriptor)
- {
+ for (int i = 0; i < LIBAIO_QUEUE_SIZE / 2; i++) {
+ for (LibaioFile file : fileDescriptor) {
file.write(i * 512, 512, buffer, callback);
}
}
Assert.assertEquals(LIBAIO_QUEUE_SIZE, control.poll(callbacks, LIBAIO_QUEUE_SIZE, LIBAIO_QUEUE_SIZE));
- for (Object returnedCallback : callbacks)
- {
+ for (Object returnedCallback : callbacks) {
Assert.assertSame(returnedCallback, callback);
}
- for (LibaioFile file : fileDescriptor)
- {
+ for (LibaioFile file : fileDescriptor) {
ByteBuffer bigbuffer = LibaioContext.newAlignedBuffer(512 * 25, 512);
file.read(0, 512 * 25, bigbuffer, callback);
Assert.assertEquals(1, control.poll(callbacks, 1, LIBAIO_QUEUE_SIZE));
- for (Object returnedCallback : callbacks)
- {
+ for (Object returnedCallback : callbacks) {
Assert.assertSame(returnedCallback, callback);
}
- for (int i = 0; i < 512 * 25; i++)
- {
+ for (int i = 0; i < 512 * 25; i++) {
Assert.assertEquals((byte) 'a', bigbuffer.get());
}
@@ -197,15 +178,13 @@ public class LibaioTest
file.close();
}
}
- finally
- {
+ finally {
LibaioContext.freeBuffer(buffer);
}
}
@Test
- public void testSubmitWriteAndRead() throws Exception
- {
+ public void testSubmitWriteAndRead() throws Exception {
TestInfo callback = new TestInfo();
TestInfo[] callbacks = new TestInfo[LIBAIO_QUEUE_SIZE];
@@ -215,10 +194,8 @@ public class LibaioTest
// ByteBuffer buffer = ByteBuffer.allocateDirect(512);
ByteBuffer buffer = LibaioContext.newAlignedBuffer(512, 512);
- try
- {
- for (int i = 0; i < 512; i++)
- {
+ try {
+ for (int i = 0; i < 512; i++) {
buffer.put((byte) 'a');
}
@@ -235,8 +212,7 @@ public class LibaioTest
buffer = LibaioContext.newAlignedBuffer(512, 512);
- for (int i = 0; i < 512; i++)
- {
+ for (int i = 0; i < 512; i++) {
buffer.put((byte) 'B');
}
@@ -250,13 +226,11 @@ public class LibaioTest
Assert.assertEquals(1, control.poll(callbacks, 1, LIBAIO_QUEUE_SIZE));
- for (int i = 0; i < 512; i++)
- {
+ for (int i = 0; i < 512; i++) {
Assert.assertEquals('B', buffer.get());
}
}
- finally
- {
+ finally {
LibaioContext.freeBuffer(buffer);
fileDescriptor.close();
}
@@ -266,9 +240,7 @@ public class LibaioTest
/**
* This file is making use of libaio without O_DIRECT
* We won't need special buffers on this case.
- */
- public void testSubmitWriteAndReadRegularBuffers() throws Exception
- {
+ */ public void testSubmitWriteAndReadRegularBuffers() throws Exception {
TestInfo callback = new TestInfo();
TestInfo[] callbacks = new TestInfo[LIBAIO_QUEUE_SIZE];
@@ -283,10 +255,8 @@ public class LibaioTest
ByteBuffer buffer = ByteBuffer.allocateDirect(BUFFER_SIZE);
- try
- {
- for (int i = 0; i < BUFFER_SIZE; i++)
- {
+ try {
+ for (int i = 0; i < BUFFER_SIZE; i++) {
buffer.put((byte) 'a');
}
@@ -302,8 +272,7 @@ public class LibaioTest
buffer.rewind();
- for (int i = 0; i < BUFFER_SIZE; i++)
- {
+ for (int i = 0; i < BUFFER_SIZE; i++) {
buffer.put((byte) 'B');
}
@@ -317,20 +286,17 @@ public class LibaioTest
Assert.assertEquals(1, control.poll(callbacks, 1, LIBAIO_QUEUE_SIZE));
- for (int i = 0; i < BUFFER_SIZE; i++)
- {
+ for (int i = 0; i < BUFFER_SIZE; i++) {
Assert.assertEquals('B', buffer.get());
}
}
- finally
- {
+ finally {
fileDescriptor.close();
}
}
@Test
- public void testSubmitRead() throws Exception
- {
+ public void testSubmitRead() throws Exception {
TestInfo callback = new TestInfo();
@@ -345,10 +311,8 @@ public class LibaioTest
ByteBuffer buffer = LibaioContext.newAlignedBuffer(512, 512);
final int BUFFER_SIZE = 512;
- try
- {
- for (int i = 0; i < BUFFER_SIZE; i++)
- {
+ try {
+ for (int i = 0; i < BUFFER_SIZE; i++) {
buffer.put((byte) '@');
}
@@ -364,21 +328,18 @@ public class LibaioTest
Assert.assertSame(callback, callbacks[0]);
- for (int i = 0; i < BUFFER_SIZE; i++)
- {
+ for (int i = 0; i < BUFFER_SIZE; i++) {
Assert.assertEquals('@', buffer.get());
}
}
- finally
- {
+ finally {
LibaioContext.freeBuffer(buffer);
fileDescriptor.close();
}
}
@Test
- public void testInvalidWrite() throws Exception
- {
+ public void testInvalidWrite() throws Exception {
TestInfo callback = new TestInfo();
@@ -390,11 +351,9 @@ public class LibaioTest
LibaioFile fileDescriptor = control.openFile(file, true);
- try
- {
+ try {
ByteBuffer buffer = ByteBuffer.allocateDirect(300);
- for (int i = 0; i < 300; i++)
- {
+ for (int i = 0; i < 300; i++) {
buffer.put((byte) 'z');
}
@@ -410,8 +369,7 @@ public class LibaioTest
System.out.println("Error:" + callbacks[0]);
buffer = fileDescriptor.newBuffer(512);
- for (int i = 0; i < 512; i++)
- {
+ for (int i = 0; i < 512; i++) {
buffer.put((byte) 'z');
}
@@ -434,15 +392,13 @@ public class LibaioTest
TestInfo.checkLeaks();
}
- finally
- {
+ finally {
fileDescriptor.close();
}
}
@Test
- public void testLeaks() throws Exception
- {
+ public void testLeaks() throws Exception {
File file = temporaryFolder.newFile("test.bin");
fillupFile(file, LIBAIO_QUEUE_SIZE * 2);
@@ -453,25 +409,20 @@ public class LibaioTest
ByteBuffer bufferWrite = LibaioContext.newAlignedBuffer(512, 512);
- try
- {
- for (int i = 0; i < 512; i++)
- {
+ try {
+ for (int i = 0; i < 512; i++) {
bufferWrite.put((byte) 'B');
}
- for (int j = 0; j < LIBAIO_QUEUE_SIZE * 2; j++)
- {
- for (int i = 0; i < LIBAIO_QUEUE_SIZE; i++)
- {
+ for (int j = 0; j < LIBAIO_QUEUE_SIZE * 2; j++) {
+ for (int i = 0; i < LIBAIO_QUEUE_SIZE; i++) {
TestInfo countClass = new TestInfo();
fileDescriptor.write(i * 512, 512, bufferWrite, countClass);
}
Assert.assertEquals(LIBAIO_QUEUE_SIZE, control.poll(callbacks, LIBAIO_QUEUE_SIZE, LIBAIO_QUEUE_SIZE));
- for (int i = 0; i < LIBAIO_QUEUE_SIZE; i++)
- {
+ for (int i = 0; i < LIBAIO_QUEUE_SIZE; i++) {
Assert.assertNotNull(callbacks[i]);
callbacks[i] = null;
}
@@ -479,15 +430,13 @@ public class LibaioTest
TestInfo.checkLeaks();
}
- finally
- {
+ finally {
LibaioContext.freeBuffer(bufferWrite);
}
}
@Test
- public void testLock() throws Exception
- {
+ public void testLock() throws Exception {
File file = temporaryFolder.newFile("test.bin");
LibaioFile fileDescriptor = control.openFile(file, true);
@@ -497,8 +446,7 @@ public class LibaioTest
}
@Test
- public void testAlloc() throws Exception
- {
+ public void testAlloc() throws Exception {
File file = temporaryFolder.newFile("test.bin");
LibaioFile fileDescriptor = control.openFile(file, true);
@@ -508,15 +456,12 @@ public class LibaioTest
}
@Test
- public void testReleaseNullBuffer() throws Exception
- {
+ public void testReleaseNullBuffer() throws Exception {
boolean failed = false;
- try
- {
+ try {
LibaioContext.freeBuffer(null);
}
- catch (Exception expected)
- {
+ catch (Exception expected) {
failed = true;
}
@@ -525,20 +470,17 @@ public class LibaioTest
}
@Test
- public void testMemset() throws Exception
- {
+ public void testMemset() throws Exception {
ByteBuffer buffer = LibaioContext.newAlignedBuffer(512 * 8, 512);
- for (int i = 0; i < buffer.capacity(); i++)
- {
+ for (int i = 0; i < buffer.capacity(); i++) {
buffer.put((byte) 'z');
}
buffer.position(0);
- for (int i = 0; i < buffer.capacity(); i++)
- {
+ for (int i = 0; i < buffer.capacity(); i++) {
Assert.assertEquals((byte) 'z', buffer.get());
}
@@ -546,8 +488,7 @@ public class LibaioTest
buffer.position(0);
- for (int i = 0; i < buffer.capacity(); i++)
- {
+ for (int i = 0; i < buffer.capacity(); i++) {
Assert.assertEquals((byte) 0, buffer.get());
}
@@ -556,32 +497,27 @@ public class LibaioTest
}
@Test
- public void testIOExceptionConditions() throws Exception
- {
+ public void testIOExceptionConditions() throws Exception {
boolean exceptionThrown = false;
control.close();
control = new LibaioContext<>(LIBAIO_QUEUE_SIZE, false);
- try
- {
+ try {
// There is no space for a queue this huge, the native layer should throw the exception
LibaioContext newController = new LibaioContext(Integer.MAX_VALUE, false);
}
- catch (RuntimeException e)
- {
+ catch (RuntimeException e) {
exceptionThrown = true;
}
Assert.assertTrue(exceptionThrown);
exceptionThrown = false;
- try
- {
+ try {
// this should throw an exception, we shouldn't be able to open a directory!
control.openFile(temporaryFolder.getRoot(), true);
}
- catch (IOException expected)
- {
+ catch (IOException expected) {
exceptionThrown = true;
}
@@ -591,12 +527,10 @@ public class LibaioTest
LibaioFile fileDescriptor = control.openFile(temporaryFolder.newFile(), true);
fileDescriptor.close();
- try
- {
+ try {
fileDescriptor.close();
}
- catch (IOException expected)
- {
+ catch (IOException expected) {
exceptionThrown = true;
}
@@ -606,25 +540,20 @@ public class LibaioTest
ByteBuffer buffer = fileDescriptor.newBuffer(512);
- try
- {
- for (int i = 0; i < 512; i++)
- {
+ try {
+ for (int i = 0; i < 512; i++) {
buffer.put((byte) 'a');
}
- for (int i = 0; i < LIBAIO_QUEUE_SIZE; i++)
- {
+ for (int i = 0; i < LIBAIO_QUEUE_SIZE; i++) {
fileDescriptor.write(i * 512, 512, buffer, new TestInfo());
}
boolean ex = false;
- try
- {
+ try {
fileDescriptor.write(0, 512, buffer, new TestInfo());
}
- catch (Exception e)
- {
+ catch (Exception e) {
ex = true;
}
@@ -651,50 +580,41 @@ public class LibaioTest
TestInfo.checkLeaks();
exceptionThrown = false;
- try
- {
+ try {
LibaioContext.newAlignedBuffer(300, 512);
}
- catch (RuntimeException e)
- {
+ catch (RuntimeException e) {
exceptionThrown = true;
}
Assert.assertTrue(exceptionThrown);
exceptionThrown = false;
- try
- {
+ try {
LibaioContext.newAlignedBuffer(-512, 512);
}
- catch (RuntimeException e)
- {
+ catch (RuntimeException e) {
exceptionThrown = true;
}
Assert.assertTrue(exceptionThrown);
}
- finally
- {
+ finally {
LibaioContext.freeBuffer(buffer);
}
}
@Test
- public void testBlockedCallback() throws Exception
- {
+ public void testBlockedCallback() throws Exception {
final LibaioContext blockedContext = new LibaioContext(500, true);
- Thread t = new Thread()
- {
- public void run()
- {
+ Thread t = new Thread() {
+ public void run() {
blockedContext.poll();
}
};
t.start();
-
int NUMBER_OF_BLOCKS = 5000;
final CountDownLatch latch = new CountDownLatch(NUMBER_OF_BLOCKS);
@@ -705,17 +625,15 @@ public class LibaioTest
final AtomicInteger errors = new AtomicInteger(0);
- class MyCallback implements SubmitInfo
- {
+ class MyCallback implements SubmitInfo {
+
@Override
- public void onError(int errno, String message)
- {
+ public void onError(int errno, String message) {
errors.incrementAndGet();
}
@Override
- public void done()
- {
+ public void done() {
latch.countDown();
}
}
@@ -724,16 +642,13 @@ public class LibaioTest
ByteBuffer buffer = LibaioContext.newAlignedBuffer(512, 512);
-
- for (int i = 0; i < 512; i++)
- {
- buffer.put((byte)'a');
+ for (int i = 0; i < 512; i++) {
+ buffer.put((byte) 'a');
}
long start = System.currentTimeMillis();
- for (int i = 0; i < NUMBER_OF_BLOCKS; i++)
- {
+ for (int i = 0; i < NUMBER_OF_BLOCKS; i++) {
aioFile.write(i * 512, 512, buffer, callback);
}
@@ -741,19 +656,18 @@ public class LibaioTest
latch.await();
-
System.out.println("time = " + (end - start) + " writes/second=" + NUMBER_OF_BLOCKS * 1000L / (end - start));
-//
-// MultiThreadAsynchronousFileTest.debug((sync ? "Sync result:" : "Async result:") + " Records/Second = " +
-// MultiThreadAsynchronousFileTest.NUMBER_OF_THREADS *
-// MultiThreadAsynchronousFileTest.NUMBER_OF_LINES *
-// 1000 /
-// (endTime - startTime) +
-// " total time = " +
-// (endTime - startTime) +
-// " total number of records = " +
-// MultiThreadAsynchronousFileTest.NUMBER_OF_THREADS *
-// MultiThreadAsynchronousFileTest.NUMBER_OF_LINES);
+ //
+ // MultiThreadAsynchronousFileTest.debug((sync ? "Sync result:" : "Async result:") + " Records/Second = " +
+ // MultiThreadAsynchronousFileTest.NUMBER_OF_THREADS *
+ // MultiThreadAsynchronousFileTest.NUMBER_OF_LINES *
+ // 1000 /
+ // (endTime - startTime) +
+ // " total time = " +
+ // (endTime - startTime) +
+ // " total number of records = " +
+ // MultiThreadAsynchronousFileTest.NUMBER_OF_THREADS *
+ // MultiThreadAsynchronousFileTest.NUMBER_OF_LINES);
Thread.sleep(100);
@@ -761,42 +675,34 @@ public class LibaioTest
t.join();
}
- private void fillupFile(File file, int blocks) throws IOException
- {
+ private void fillupFile(File file, int blocks) throws IOException {
FileOutputStream fileOutputStream = new FileOutputStream(file);
byte[] bufferWrite = new byte[512];
- for (int i = 0; i < 512; i++)
- {
+ for (int i = 0; i < 512; i++) {
bufferWrite[i] = (byte) 0;
}
- for (int i = 0; i < blocks; i++)
- {
+ for (int i = 0; i < blocks; i++) {
fileOutputStream.write(bufferWrite);
}
fileOutputStream.close();
}
+ static class TestInfo implements SubmitInfo {
- static class TestInfo implements SubmitInfo
- {
static AtomicInteger count = new AtomicInteger();
@Override
- protected void finalize() throws Throwable
- {
+ protected void finalize() throws Throwable {
super.finalize();
count.decrementAndGet();
}
- public static void checkLeaks() throws InterruptedException
- {
- for (int i = 0; count.get() != 0 && i < 50; i++)
- {
+ public static void checkLeaks() throws InterruptedException {
+ for (int i = 0; count.get() != 0 && i < 50; i++) {
WeakReference reference = new WeakReference(new Object());
- while (reference.get() != null)
- {
+ while (reference.get() != null) {
System.gc();
Thread.sleep(100);
}
@@ -808,51 +714,42 @@ public class LibaioTest
String errorMessage;
int errno;
- public TestInfo()
- {
+ public TestInfo() {
count.incrementAndGet();
}
@Override
- public void onError(int errno, String message)
- {
+ public void onError(int errno, String message) {
this.errno = errno;
this.errorMessage = message;
this.error = true;
}
@Override
- public void done()
- {
+ public void done() {
}
- public int getErrno()
- {
+ public int getErrno() {
return errno;
}
- public void setErrno(int errno)
- {
+ public void setErrno(int errno) {
this.errno = errno;
}
- public boolean isError()
- {
+ public boolean isError() {
return error;
}
- public void setError(boolean error)
- {
+ public void setError(boolean error) {
this.error = error;
}
- public String getErrorMessage()
- {
+ public String getErrorMessage() {
return errorMessage;
}
- public void setErrorMessage(String errorMessage)
- {
+ public void setErrorMessage(String errorMessage) {
this.errorMessage = errorMessage;
}
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/bac96047/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/ActiveMQProtonRemotingConnection.java
----------------------------------------------------------------------
diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/ActiveMQProtonRemotingConnection.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/ActiveMQProtonRemotingConnection.java
index 3b8ca6a..ea8582a 100644
--- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/ActiveMQProtonRemotingConnection.java
+++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/ActiveMQProtonRemotingConnection.java
@@ -28,39 +28,36 @@ import org.proton.plug.AMQPConnectionContext;
/**
* This is a Server's Connection representation used by ActiveMQ Artemis.
*/
-public class ActiveMQProtonRemotingConnection extends AbstractRemotingConnection
-{
+public class ActiveMQProtonRemotingConnection extends AbstractRemotingConnection {
+
private final AMQPConnectionContext amqpConnection;
private boolean destroyed = false;
private final ProtonProtocolManager manager;
-
- public ActiveMQProtonRemotingConnection(ProtonProtocolManager manager, AMQPConnectionContext amqpConnection, Connection transportConnection, Executor executor)
- {
+ public ActiveMQProtonRemotingConnection(ProtonProtocolManager manager,
+ AMQPConnectionContext amqpConnection,
+ Connection transportConnection,
+ Executor executor) {
super(transportConnection, executor);
this.manager = manager;
this.amqpConnection = amqpConnection;
}
- public Executor getExecutor()
- {
+ public Executor getExecutor() {
return this.executor;
}
- public ProtonProtocolManager getManager()
- {
+ public ProtonProtocolManager getManager() {
return manager;
}
/*
* This can be called concurrently by more than one thread so needs to be locked
*/
- public void fail(final ActiveMQException me, String scaleDownTargetNodeID)
- {
- if (destroyed)
- {
+ public void fail(final ActiveMQException me, String scaleDownTargetNodeID) {
+ if (destroyed) {
return;
}
@@ -76,21 +73,16 @@ public class ActiveMQProtonRemotingConnection extends AbstractRemotingConnection
internalClose();
}
-
@Override
- public void destroy()
- {
- synchronized (this)
- {
- if (destroyed)
- {
+ public void destroy() {
+ synchronized (this) {
+ if (destroyed) {
return;
}
destroyed = true;
}
-
callClosingListeners();
internalClose();
@@ -98,20 +90,17 @@ public class ActiveMQProtonRemotingConnection extends AbstractRemotingConnection
}
@Override
- public boolean isClient()
- {
+ public boolean isClient() {
return false;
}
@Override
- public boolean isDestroyed()
- {
+ public boolean isDestroyed() {
return destroyed;
}
@Override
- public void disconnect(boolean criticalError)
- {
+ public void disconnect(boolean criticalError) {
getTransportConnection().close();
}
@@ -119,32 +108,27 @@ public class ActiveMQProtonRemotingConnection extends AbstractRemotingConnection
* Disconnect the connection, closing all channels
*/
@Override
- public void disconnect(String scaleDownNodeID, boolean criticalError)
- {
+ public void disconnect(String scaleDownNodeID, boolean criticalError) {
getTransportConnection().close();
}
@Override
- public boolean checkDataReceived()
- {
+ public boolean checkDataReceived() {
return amqpConnection.checkDataReceived();
}
@Override
- public void flush()
- {
+ public void flush() {
amqpConnection.flush();
}
@Override
- public void bufferReceived(Object connectionID, ActiveMQBuffer buffer)
- {
+ public void bufferReceived(Object connectionID, ActiveMQBuffer buffer) {
amqpConnection.inputBuffer(buffer.byteBuf());
super.bufferReceived(connectionID, buffer);
}
- private void internalClose()
- {
+ private void internalClose() {
// We close the underlying transport connection
getTransportConnection().close();
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/bac96047/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/ProtonProtocolManager.java
----------------------------------------------------------------------
diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/ProtonProtocolManager.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/ProtonProtocolManager.java
index f50d3e4..3637289 100644
--- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/ProtonProtocolManager.java
+++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/ProtonProtocolManager.java
@@ -46,67 +46,54 @@ import static org.proton.plug.context.AMQPConstants.Connection.DEFAULT_MAX_FRAME
/**
* A proton protocol manager, basically reads the Proton Input and maps proton resources to ActiveMQ Artemis resources
*/
-public class ProtonProtocolManager implements ProtocolManager<Interceptor>, NotificationListener
-{
+public class ProtonProtocolManager implements ProtocolManager<Interceptor>, NotificationListener {
+
private final ActiveMQServer server;
private MessageConverter protonConverter;
-
private final ProtonProtocolManagerFactory factory;
- public ProtonProtocolManager(ProtonProtocolManagerFactory factory, ActiveMQServer server)
- {
+ public ProtonProtocolManager(ProtonProtocolManagerFactory factory, ActiveMQServer server) {
this.factory = factory;
this.server = server;
this.protonConverter = new ProtonMessageConverter(server.getStorageManager());
}
- public ActiveMQServer getServer()
- {
+ public ActiveMQServer getServer() {
return server;
}
-
@Override
- public MessageConverter getConverter()
- {
+ public MessageConverter getConverter() {
return protonConverter;
}
@Override
- public void onNotification(Notification notification)
- {
+ public void onNotification(Notification notification) {
}
@Override
- public ProtocolManagerFactory<Interceptor> getFactory()
- {
+ public ProtocolManagerFactory<Interceptor> getFactory() {
return factory;
}
@Override
- public void updateInterceptors(List<BaseInterceptor> incomingInterceptors, List<BaseInterceptor> outgoingInterceptors)
- {
+ public void updateInterceptors(List<BaseInterceptor> incomingInterceptors,
+ List<BaseInterceptor> outgoingInterceptors) {
// no op
}
@Override
- public ConnectionEntry createConnectionEntry(Acceptor acceptorUsed, Connection remotingConnection)
- {
+ public ConnectionEntry createConnectionEntry(Acceptor acceptorUsed, Connection remotingConnection) {
ActiveMQProtonConnectionCallback connectionCallback = new ActiveMQProtonConnectionCallback(this, remotingConnection);
long ttl = ActiveMQClient.DEFAULT_CONNECTION_TTL;
- if (server.getConfiguration().getConnectionTTLOverride() != -1)
- {
+ if (server.getConfiguration().getConnectionTTLOverride() != -1) {
ttl = server.getConfiguration().getConnectionTTLOverride();
}
- AMQPServerConnectionContext amqpConnection = ProtonServerConnectionContextFactory.getFactory().createConnection(
- connectionCallback,
- (int) ttl,
- DEFAULT_MAX_FRAME_SIZE,
- DEFAULT_CHANNEL_MAX);
+ AMQPServerConnectionContext amqpConnection = ProtonServerConnectionContextFactory.getFactory().createConnection(connectionCallback, (int) ttl, DEFAULT_MAX_FRAME_SIZE, DEFAULT_CHANNEL_MAX);
Executor executor = server.getExecutorFactory().getExecutor();
@@ -114,42 +101,35 @@ public class ProtonProtocolManager implements ProtocolManager<Interceptor>, Noti
connectionCallback.setProtonConnectionDelegate(delegate);
- ConnectionEntry entry = new ConnectionEntry(delegate, executor,
- System.currentTimeMillis(), ttl);
+ ConnectionEntry entry = new ConnectionEntry(delegate, executor, System.currentTimeMillis(), ttl);
return entry;
}
@Override
- public void removeHandler(String name)
- {
+ public void removeHandler(String name) {
}
@Override
- public void handleBuffer(RemotingConnection connection, ActiveMQBuffer buffer)
- {
+ public void handleBuffer(RemotingConnection connection, ActiveMQBuffer buffer) {
ActiveMQProtonRemotingConnection protonConnection = (ActiveMQProtonRemotingConnection) connection;
protonConnection.bufferReceived(protonConnection.getID(), buffer);
}
@Override
- public void addChannelHandlers(ChannelPipeline pipeline)
- {
+ public void addChannelHandlers(ChannelPipeline pipeline) {
}
@Override
- public boolean isProtocol(byte[] array)
- {
+ public boolean isProtocol(byte[] array) {
return array.length >= 4 && array[0] == (byte) 'A' && array[1] == (byte) 'M' && array[2] == (byte) 'Q' && array[3] == (byte) 'P';
}
@Override
- public void handshake(NettyServerConnection connection, ActiveMQBuffer buffer)
- {
+ public void handshake(NettyServerConnection connection, ActiveMQBuffer buffer) {
}
-
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/bac96047/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/ProtonProtocolManagerFactory.java
----------------------------------------------------------------------
diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/ProtonProtocolManagerFactory.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/ProtonProtocolManagerFactory.java
index f45daae..7a1d25e 100644
--- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/ProtonProtocolManagerFactory.java
+++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/ProtonProtocolManagerFactory.java
@@ -25,8 +25,8 @@ import org.apache.activemq.artemis.spi.core.protocol.ProtocolManager;
import java.util.Collections;
import java.util.List;
-public class ProtonProtocolManagerFactory extends AbstractProtocolManagerFactory<Interceptor>
-{
+public class ProtonProtocolManagerFactory extends AbstractProtocolManagerFactory<Interceptor> {
+
private static final String AMQP_PROTOCOL_NAME = "AMQP";
private static final String MODULE_NAME = "artemis-amqp-protocol";
@@ -34,27 +34,25 @@ public class ProtonProtocolManagerFactory extends AbstractProtocolManagerFactory
private static String[] SUPPORTED_PROTOCOLS = {AMQP_PROTOCOL_NAME};
@Override
- public ProtocolManager createProtocolManager(ActiveMQServer server, List<Interceptor> incomingInterceptors, List<Interceptor> outgoingInterceptors)
- {
+ public ProtocolManager createProtocolManager(ActiveMQServer server,
+ List<Interceptor> incomingInterceptors,
+ List<Interceptor> outgoingInterceptors) {
return new ProtonProtocolManager(this, server);
}
@Override
- public List<Interceptor> filterInterceptors(List<BaseInterceptor> interceptors)
- {
+ public List<Interceptor> filterInterceptors(List<BaseInterceptor> interceptors) {
// no interceptors on Proton
return Collections.emptyList();
}
@Override
- public String[] getProtocols()
- {
+ public String[] getProtocols() {
return SUPPORTED_PROTOCOLS;
}
@Override
- public String getModuleName()
- {
+ public String getModuleName() {
return MODULE_NAME;
}
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/bac96047/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/converter/ActiveMQJMSVendor.java
----------------------------------------------------------------------
diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/converter/ActiveMQJMSVendor.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/converter/ActiveMQJMSVendor.java
index 606cf31..8e7502b 100644
--- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/converter/ActiveMQJMSVendor.java
+++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/converter/ActiveMQJMSVendor.java
@@ -35,92 +35,75 @@ import org.apache.activemq.artemis.core.server.ServerMessage;
import org.apache.activemq.artemis.core.server.impl.ServerMessageImpl;
import org.apache.activemq.artemis.utils.IDGenerator;
-public class ActiveMQJMSVendor extends JMSVendor
-{
+public class ActiveMQJMSVendor extends JMSVendor {
private final IDGenerator serverGenerator;
- ActiveMQJMSVendor(IDGenerator idGenerator)
- {
+ ActiveMQJMSVendor(IDGenerator idGenerator) {
this.serverGenerator = idGenerator;
}
@Override
- public BytesMessage createBytesMessage()
- {
+ public BytesMessage createBytesMessage() {
return new ServerJMSBytesMessage(newMessage(org.apache.activemq.artemis.api.core.Message.BYTES_TYPE), 0);
}
@Override
- public StreamMessage createStreamMessage()
- {
+ public StreamMessage createStreamMessage() {
return new ServerJMSStreamMessage(newMessage(org.apache.activemq.artemis.api.core.Message.STREAM_TYPE), 0);
}
@Override
- public Message createMessage()
- {
- return new ServerJMSMessage(newMessage(org.apache.activemq.artemis.api.core.Message.DEFAULT_TYPE), 0 );
+ public Message createMessage() {
+ return new ServerJMSMessage(newMessage(org.apache.activemq.artemis.api.core.Message.DEFAULT_TYPE), 0);
}
@Override
- public TextMessage createTextMessage()
- {
+ public TextMessage createTextMessage() {
return new ServerJMSTextMessage(newMessage(org.apache.activemq.artemis.api.core.Message.TEXT_TYPE), 0);
}
@Override
- public ObjectMessage createObjectMessage()
- {
+ public ObjectMessage createObjectMessage() {
return null;
}
@Override
- public MapMessage createMapMessage()
- {
+ public MapMessage createMapMessage() {
return new ServerJMSMapMessage(newMessage(org.apache.activemq.artemis.api.core.Message.MAP_TYPE), 0);
}
@Override
- public void setJMSXUserID(Message message, String s)
- {
+ public void setJMSXUserID(Message message, String s) {
}
@Override
- public Destination createDestination(String name)
- {
+ public Destination createDestination(String name) {
return super.createDestination(name);
}
@Override
- public <T extends Destination> T createDestination(String name, Class<T> kind)
- {
+ public <T extends Destination> T createDestination(String name, Class<T> kind) {
return super.createDestination(name, kind);
}
@Override
- public void setJMSXGroupID(Message message, String s)
- {
+ public void setJMSXGroupID(Message message, String s) {
}
@Override
- public void setJMSXGroupSequence(Message message, int i)
- {
+ public void setJMSXGroupSequence(Message message, int i) {
}
@Override
- public void setJMSXDeliveryCount(Message message, long l)
- {
+ public void setJMSXDeliveryCount(Message message, long l) {
}
-
- public ServerJMSMessage wrapMessage(int messageType, ServerMessage wrapped, int deliveryCount)
- {
- switch (messageType)
- {
+ public ServerJMSMessage wrapMessage(int messageType, ServerMessage wrapped, int deliveryCount) {
+ switch (messageType) {
case org.apache.activemq.artemis.api.core.Message.STREAM_TYPE:
return new ServerJMSStreamMessage(wrapped, deliveryCount);
case org.apache.activemq.artemis.api.core.Message.BYTES_TYPE:
@@ -135,19 +118,15 @@ public class ActiveMQJMSVendor extends JMSVendor
}
-
@Override
- public String toAddress(Destination destination)
- {
+ public String toAddress(Destination destination) {
return null;
}
-
- private ServerMessageImpl newMessage(byte messageType)
- {
+ private ServerMessageImpl newMessage(byte messageType) {
ServerMessageImpl message = new ServerMessageImpl(serverGenerator.generateID(), 512);
message.setType(messageType);
- ((ResetLimitWrappedActiveMQBuffer)message.getBodyBuffer()).setMessage(null);
+ ((ResetLimitWrappedActiveMQBuffer) message.getBodyBuffer()).setMessage(null);
return message;
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/bac96047/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/converter/ProtonMessageConverter.java
----------------------------------------------------------------------
diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/converter/ProtonMessageConverter.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/converter/ProtonMessageConverter.java
index 5cee473..4de2357 100644
--- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/converter/ProtonMessageConverter.java
+++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/converter/ProtonMessageConverter.java
@@ -25,14 +25,11 @@ import org.apache.activemq.artemis.core.server.ServerMessage;
import org.apache.activemq.artemis.spi.core.protocol.MessageConverter;
import org.apache.activemq.artemis.utils.IDGenerator;
-public class ProtonMessageConverter implements MessageConverter
-{
-
+public class ProtonMessageConverter implements MessageConverter {
ActiveMQJMSVendor activeMQJMSVendor;
- public ProtonMessageConverter(IDGenerator idGenerator)
- {
+ public ProtonMessageConverter(IDGenerator idGenerator) {
activeMQJMSVendor = new ActiveMQJMSVendor(idGenerator);
inboundTransformer = new JMSMappingInboundTransformer(activeMQJMSVendor);
outboundTransformer = new JMSMappingOutboundTransformer(activeMQJMSVendor);
@@ -42,33 +39,30 @@ public class ProtonMessageConverter implements MessageConverter
private final JMSMappingOutboundTransformer outboundTransformer;
@Override
- public ServerMessage inbound(Object messageSource) throws Exception
- {
+ public ServerMessage inbound(Object messageSource) throws Exception {
ServerJMSMessage jmsMessage = inboundJMSType((EncodedMessage) messageSource);
- return (ServerMessage)jmsMessage.getInnerMessage();
+ return (ServerMessage) jmsMessage.getInnerMessage();
}
/**
* Just create the JMS Part of the inbound (for testing)
+ *
* @param messageSource
* @return
* @throws Exception
*/
- public ServerJMSMessage inboundJMSType(EncodedMessage messageSource) throws Exception
- {
+ public ServerJMSMessage inboundJMSType(EncodedMessage messageSource) throws Exception {
EncodedMessage encodedMessageSource = messageSource;
- ServerJMSMessage transformedMessage = (ServerJMSMessage)inboundTransformer.transform(encodedMessageSource);
+ ServerJMSMessage transformedMessage = (ServerJMSMessage) inboundTransformer.transform(encodedMessageSource);
transformedMessage.encode();
return transformedMessage;
}
-
@Override
- public Object outbound(ServerMessage messageOutbound, int deliveryCount) throws Exception
- {
+ public Object outbound(ServerMessage messageOutbound, int deliveryCount) throws Exception {
ServerJMSMessage jmsMessage = activeMQJMSVendor.wrapMessage(messageOutbound.getType(), messageOutbound, deliveryCount);
jmsMessage.decode();
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/bac96047/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/converter/jms/ServerJMSBytesMessage.java
----------------------------------------------------------------------
diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/converter/jms/ServerJMSBytesMessage.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/converter/jms/ServerJMSBytesMessage.java
index 4389c66..76e2515 100644
--- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/converter/jms/ServerJMSBytesMessage.java
+++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/converter/jms/ServerJMSBytesMessage.java
@@ -47,190 +47,158 @@ import static org.apache.activemq.artemis.reader.BytesMessageUtil.bytesWriteObje
import static org.apache.activemq.artemis.reader.BytesMessageUtil.bytesWriteShort;
import static org.apache.activemq.artemis.reader.BytesMessageUtil.bytesWriteUTF;
-public class ServerJMSBytesMessage extends ServerJMSMessage implements BytesMessage
-{
- public ServerJMSBytesMessage(MessageInternal message, int deliveryCount)
- {
+public class ServerJMSBytesMessage extends ServerJMSMessage implements BytesMessage {
+
+ public ServerJMSBytesMessage(MessageInternal message, int deliveryCount) {
super(message, deliveryCount);
}
@Override
- public long getBodyLength() throws JMSException
- {
+ public long getBodyLength() throws JMSException {
return message.getEndOfBodyPosition() - MessageImpl.BODY_OFFSET;
}
@Override
- public boolean readBoolean() throws JMSException
- {
+ public boolean readBoolean() throws JMSException {
return bytesReadBoolean(message);
}
@Override
- public byte readByte() throws JMSException
- {
+ public byte readByte() throws JMSException {
return bytesReadByte(message);
}
@Override
- public int readUnsignedByte() throws JMSException
- {
+ public int readUnsignedByte() throws JMSException {
return bytesReadUnsignedByte(message);
}
@Override
- public short readShort() throws JMSException
- {
+ public short readShort() throws JMSException {
return bytesReadShort(message);
}
@Override
- public int readUnsignedShort() throws JMSException
- {
+ public int readUnsignedShort() throws JMSException {
return bytesReadUnsignedShort(message);
}
@Override
- public char readChar() throws JMSException
- {
+ public char readChar() throws JMSException {
return bytesReadChar(message);
}
@Override
- public int readInt() throws JMSException
- {
+ public int readInt() throws JMSException {
return bytesReadInt(message);
}
@Override
- public long readLong() throws JMSException
- {
+ public long readLong() throws JMSException {
return bytesReadLong(message);
}
@Override
- public float readFloat() throws JMSException
- {
+ public float readFloat() throws JMSException {
return bytesReadFloat(message);
}
@Override
- public double readDouble() throws JMSException
- {
+ public double readDouble() throws JMSException {
return bytesReadDouble(message);
}
@Override
- public String readUTF() throws JMSException
- {
+ public String readUTF() throws JMSException {
return bytesReadUTF(message);
}
@Override
- public int readBytes(byte[] value) throws JMSException
- {
+ public int readBytes(byte[] value) throws JMSException {
return bytesReadBytes(message, value);
}
@Override
- public int readBytes(byte[] value, int length) throws JMSException
- {
+ public int readBytes(byte[] value, int length) throws JMSException {
return bytesReadBytes(message, value, length);
}
@Override
- public void writeBoolean(boolean value) throws JMSException
- {
+ public void writeBoolean(boolean value) throws JMSException {
bytesWriteBoolean(message, value);
}
@Override
- public void writeByte(byte value) throws JMSException
- {
+ public void writeByte(byte value) throws JMSException {
bytesWriteByte(message, value);
}
@Override
- public void writeShort(short value) throws JMSException
- {
+ public void writeShort(short value) throws JMSException {
bytesWriteShort(message, value);
}
@Override
- public void writeChar(char value) throws JMSException
- {
+ public void writeChar(char value) throws JMSException {
bytesWriteChar(message, value);
}
@Override
- public void writeInt(int value) throws JMSException
- {
+ public void writeInt(int value) throws JMSException {
bytesWriteInt(message, value);
}
@Override
- public void writeLong(long value) throws JMSException
- {
+ public void writeLong(long value) throws JMSException {
bytesWriteLong(message, value);
}
@Override
- public void writeFloat(float value) throws JMSException
- {
+ public void writeFloat(float value) throws JMSException {
bytesWriteFloat(message, value);
}
@Override
- public void writeDouble(double value) throws JMSException
- {
+ public void writeDouble(double value) throws JMSException {
bytesWriteDouble(message, value);
}
@Override
- public void writeUTF(String value) throws JMSException
- {
+ public void writeUTF(String value) throws JMSException {
bytesWriteUTF(message, value);
}
@Override
- public void writeBytes(byte[] value) throws JMSException
- {
+ public void writeBytes(byte[] value) throws JMSException {
bytesWriteBytes(message, value);
}
@Override
- public void writeBytes(byte[] value, int offset, int length) throws JMSException
- {
+ public void writeBytes(byte[] value, int offset, int length) throws JMSException {
bytesWriteBytes(message, value, offset, length);
}
@Override
- public void writeObject(Object value) throws JMSException
- {
- if (!bytesWriteObject(message, value))
- {
+ public void writeObject(Object value) throws JMSException {
+ if (!bytesWriteObject(message, value)) {
throw new JMSException("Can't make conversion of " + value + " to any known type");
}
}
- public void encode() throws Exception
- {
+ public void encode() throws Exception {
super.encode();
// this is to make sure we encode the body-length before it's persisted
getBodyLength();
}
-
- public void decode() throws Exception
- {
+ public void decode() throws Exception {
super.decode();
}
@Override
- public void reset() throws JMSException
- {
+ public void reset() throws JMSException {
bytesMessageReset(message);
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/bac96047/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/converter/jms/ServerJMSMapMessage.java
----------------------------------------------------------------------
diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/converter/jms/ServerJMSMapMessage.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/converter/jms/ServerJMSMapMessage.java
index 9d5fb9d..bbece71 100644
--- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/converter/jms/ServerJMSMapMessage.java
+++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/converter/jms/ServerJMSMapMessage.java
@@ -36,8 +36,7 @@ import static org.apache.activemq.artemis.reader.MapMessageUtil.writeBodyMap;
/**
* ActiveMQ Artemis implementation of a JMS MapMessage.
*/
-public final class ServerJMSMapMessage extends ServerJMSMessage implements MapMessage
-{
+public final class ServerJMSMapMessage extends ServerJMSMessage implements MapMessage {
// Constants -----------------------------------------------------
public static final byte TYPE = Message.MAP_TYPE;
@@ -53,68 +52,55 @@ public final class ServerJMSMapMessage extends ServerJMSMessage implements MapMe
/*
* This constructor is used to construct messages prior to sending
*/
- public ServerJMSMapMessage(MessageInternal message, int deliveryCount)
- {
+ public ServerJMSMapMessage(MessageInternal message, int deliveryCount) {
super(message, deliveryCount);
}
// MapMessage implementation -------------------------------------
- public void setBoolean(final String name, final boolean value) throws JMSException
- {
+ public void setBoolean(final String name, final boolean value) throws JMSException {
map.putBooleanProperty(new SimpleString(name), value);
}
- public void setByte(final String name, final byte value) throws JMSException
- {
+ public void setByte(final String name, final byte value) throws JMSException {
map.putByteProperty(new SimpleString(name), value);
}
- public void setShort(final String name, final short value) throws JMSException
- {
+ public void setShort(final String name, final short value) throws JMSException {
map.putShortProperty(new SimpleString(name), value);
}
- public void setChar(final String name, final char value) throws JMSException
- {
+ public void setChar(final String name, final char value) throws JMSException {
map.putCharProperty(new SimpleString(name), value);
}
- public void setInt(final String name, final int value) throws JMSException
- {
+ public void setInt(final String name, final int value) throws JMSException {
map.putIntProperty(new SimpleString(name), value);
}
- public void setLong(final String name, final long value) throws JMSException
- {
+ public void setLong(final String name, final long value) throws JMSException {
map.putLongProperty(new SimpleString(name), value);
}
- public void setFloat(final String name, final float value) throws JMSException
- {
+ public void setFloat(final String name, final float value) throws JMSException {
map.putFloatProperty(new SimpleString(name), value);
}
- public void setDouble(final String name, final double value) throws JMSException
- {
+ public void setDouble(final String name, final double value) throws JMSException {
map.putDoubleProperty(new SimpleString(name), value);
}
- public void setString(final String name, final String value) throws JMSException
- {
+ public void setString(final String name, final String value) throws JMSException {
map.putSimpleStringProperty(new SimpleString(name), value == null ? null : new SimpleString(value));
}
- public void setBytes(final String name, final byte[] value) throws JMSException
- {
+ public void setBytes(final String name, final byte[] value) throws JMSException {
map.putBytesProperty(new SimpleString(name), value);
}
- public void setBytes(final String name, final byte[] value, final int offset, final int length) throws JMSException
- {
- if (offset + length > value.length)
- {
+ public void setBytes(final String name, final byte[] value, final int offset, final int length) throws JMSException {
+ if (offset + length > value.length) {
throw new JMSException("Invalid offset/length");
}
byte[] newBytes = new byte[length];
@@ -122,194 +108,149 @@ public final class ServerJMSMapMessage extends ServerJMSMessage implements MapMe
map.putBytesProperty(new SimpleString(name), newBytes);
}
- public void setObject(final String name, final Object value) throws JMSException
- {
- try
- {
+ public void setObject(final String name, final Object value) throws JMSException {
+ try {
TypedProperties.setObjectProperty(new SimpleString(name), value, map);
}
- catch (ActiveMQPropertyConversionException e)
- {
+ catch (ActiveMQPropertyConversionException e) {
throw new MessageFormatException(e.getMessage());
}
}
- public boolean getBoolean(final String name) throws JMSException
- {
- try
- {
+ public boolean getBoolean(final String name) throws JMSException {
+ try {
return map.getBooleanProperty(new SimpleString(name));
}
- catch (ActiveMQPropertyConversionException e)
- {
+ catch (ActiveMQPropertyConversionException e) {
throw new MessageFormatException(e.getMessage());
}
}
- public byte getByte(final String name) throws JMSException
- {
- try
- {
+ public byte getByte(final String name) throws JMSException {
+ try {
return map.getByteProperty(new SimpleString(name));
}
- catch (ActiveMQPropertyConversionException e)
- {
+ catch (ActiveMQPropertyConversionException e) {
throw new MessageFormatException(e.getMessage());
}
}
- public short getShort(final String name) throws JMSException
- {
- try
- {
+ public short getShort(final String name) throws JMSException {
+ try {
return map.getShortProperty(new SimpleString(name));
}
- catch (ActiveMQPropertyConversionException e)
- {
+ catch (ActiveMQPropertyConversionException e) {
throw new MessageFormatException(e.getMessage());
}
}
- public char getChar(final String name) throws JMSException
- {
- try
- {
+ public char getChar(final String name) throws JMSException {
+ try {
return map.getCharProperty(new SimpleString(name));
}
- catch (ActiveMQPropertyConversionException e)
- {
+ catch (ActiveMQPropertyConversionException e) {
throw new MessageFormatException(e.getMessage());
}
}
- public int getInt(final String name) throws JMSException
- {
- try
- {
+ public int getInt(final String name) throws JMSException {
+ try {
return map.getIntProperty(new SimpleString(name));
}
- catch (ActiveMQPropertyConversionException e)
- {
+ catch (ActiveMQPropertyConversionException e) {
throw new MessageFormatException(e.getMessage());
}
}
- public long getLong(final String name) throws JMSException
- {
- try
- {
+ public long getLong(final String name) throws JMSException {
+ try {
return map.getLongProperty(new SimpleString(name));
}
- catch (ActiveMQPropertyConversionException e)
- {
+ catch (ActiveMQPropertyConversionException e) {
throw new MessageFormatException(e.getMessage());
}
}
- public float getFloat(final String name) throws JMSException
- {
- try
- {
+ public float getFloat(final String name) throws JMSException {
+ try {
return map.getFloatProperty(new SimpleString(name));
}
- catch (ActiveMQPropertyConversionException e)
- {
+ catch (ActiveMQPropertyConversionException e) {
throw new MessageFormatException(e.getMessage());
}
}
- public double getDouble(final String name) throws JMSException
- {
- try
- {
+ public double getDouble(final String name) throws JMSException {
+ try {
return map.getDoubleProperty(new SimpleString(name));
}
- catch (ActiveMQPropertyConversionException e)
- {
+ catch (ActiveMQPropertyConversionException e) {
throw new MessageFormatException(e.getMessage());
}
}
- public String getString(final String name) throws JMSException
- {
- try
- {
+ public String getString(final String name) throws JMSException {
+ try {
SimpleString str = map.getSimpleStringProperty(new SimpleString(name));
- if (str == null)
- {
+ if (str == null) {
return null;
}
- else
- {
+ else {
return str.toString();
}
}
- catch (ActiveMQPropertyConversionException e)
- {
+ catch (ActiveMQPropertyConversionException e) {
throw new MessageFormatException(e.getMessage());
}
}
- public byte[] getBytes(final String name) throws JMSException
- {
- try
- {
+ public byte[] getBytes(final String name) throws JMSException {
+ try {
return map.getBytesProperty(new SimpleString(name));
}
- catch (ActiveMQPropertyConversionException e)
- {
+ catch (ActiveMQPropertyConversionException e) {
throw new MessageFormatException(e.getMessage());
}
}
- public Object getObject(final String name) throws JMSException
- {
+ public Object getObject(final String name) throws JMSException {
Object val = map.getProperty(new SimpleString(name));
- if (val instanceof SimpleString)
- {
+ if (val instanceof SimpleString) {
val = ((SimpleString) val).toString();
}
return val;
}
- public Enumeration getMapNames() throws JMSException
- {
+ public Enumeration getMapNames() throws JMSException {
Set<SimpleString> simplePropNames = map.getPropertyNames();
Set<String> propNames = new HashSet<String>(simplePropNames.size());
- for (SimpleString str : simplePropNames)
- {
+ for (SimpleString str : simplePropNames) {
propNames.add(str.toString());
}
return Collections.enumeration(propNames);
}
- public boolean itemExists(final String name) throws JMSException
- {
+ public boolean itemExists(final String name) throws JMSException {
return map.containsProperty(new SimpleString(name));
}
-
@Override
- public void clearBody() throws JMSException
- {
+ public void clearBody() throws JMSException {
super.clearBody();
map.clear();
}
-
- public void encode() throws Exception
- {
+ public void encode() throws Exception {
super.encode();
writeBodyMap(message, map);
}
- public void decode() throws Exception
- {
+ public void decode() throws Exception {
super.decode();
readBodyMap(message, map);
}