6:03 a.m.
Author: gaohoward
Date: 2010-11-11 07:03:18 -0500 (Thu, 11 Nov 2010)
New Revision: 9873
Modified:
branches/Branch_Large_Message_Compression/src/main/org/hornetq/utils/GZipUtil.java
branches/Branch_Large_Message_Compression/tests/src/org/hornetq/tests/integration/client/LargeMessageTest.java
Log:
a working compression prototype (non-thread)
Modified:
branches/Branch_Large_Message_Compression/src/main/org/hornetq/utils/GZipUtil.java
===================================================================
---
branches/Branch_Large_Message_Compression/src/main/org/hornetq/utils/GZipUtil.java 2010-11-11
02:23:43 UTC (rev 9872)
+++
branches/Branch_Large_Message_Compression/src/main/org/hornetq/utils/GZipUtil.java 2010-11-11
12:03:18 UTC (rev 9873)
@@ -15,12 +15,19 @@
import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.PipedInputStream;
import java.io.PipedOutputStream;
+import java.util.ArrayList;
+import java.util.List;
import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;
@@ -99,7 +106,7 @@
}
while ((size = input.read(readBytes)) > 0)
{
- System.out.println("Read " + size + " bytes on
compressing thread");
+// System.out.println("Read " + size + " bytes on
compressing thread");
out.write(readBytes, 0, size);
}
System.out.println("Finished compressing");
@@ -163,5 +170,212 @@
});
}
+ /*
+ * we keep a list of byte arrays. when writing, we start with the first array.
+ * when getBuffer() is called, the returned value is subject to the following rules:
+ *
+ * 1. if not closed, return the last full array. then update flags and pointers.
+ * 2. if closed, return all the remaining.
+ */
+ public static class DynamicOutputStream extends OutputStream
+ {
+ private List<byte[]> writeBuffer;
+ private int bufferSize;
+ private int counter, index;
+ private int readIndex, nextIndex;
+ private boolean closed;
+
+ public DynamicOutputStream(int size, int cache)
+ {
+ bufferSize = size;
+ writeBuffer = new ArrayList<byte[]>(cache);
+ for (int i = 0; i < cache; i++)
+ {
+ writeBuffer.add(new byte[size]);
+ }
+ counter = 0;
+ index = 0;
+ readIndex = 0;
+ nextIndex = 1;
+ closed = false;
+ }
+
+ public void write(int b) throws IOException
+ {
+ writeBuffer.get(index)[counter++] = (byte)b;
+ if (counter == bufferSize)
+ {
+ index = nextIndex;
+ nextIndex++;
+ if (index == writeBuffer.size())
+ {
+ writeBuffer.add(new byte[bufferSize]);
+ }
+ counter = 0;
+ }
+ }
+
+ public void close() throws IOException
+ {
+ closed = true;
+ }
+
+ /*
+ * logic:
+ * if index > readIndex, return readIndex, then readIndex++
+ * if index == readIndex, then return zero length byte[]. adjust nextIndex; if
closed, return the remaining.
+ * if index < readIndex, then return readIndex, readIndex = 0
+ *
+ * if closed and no more data, returns null.
+ */
+ public byte[] getBuffer()
+ {
+ byte[] result = new byte[0];
+ if (index > readIndex)
+ {
+ result = writeBuffer.get(readIndex);
+ readIndex++;
+ }
+ else if (index == readIndex)
+ {
+ if (closed)
+ {
+ if (counter == 0)
+ {
+ result = null;
+ }
+ else
+ {
+ result = new byte[counter];
+ System.arraycopy(writeBuffer.get(index), 0, result, 0, result.length);
+ counter = 0;
+ }
+ }
+ }
+ else if (index < readIndex)
+ {
+ result = writeBuffer.get(readIndex);
+ readIndex = 0;
+ }
+ return result;
+ }
+
+ }
+
+ public static class GZipPipe
+ {
+ private InputStream input;
+ private byte[] readBuffer;
+ private GZIPOutputStream zipOut;
+ private DynamicOutputStream receiver;
+
+ public GZipPipe(InputStream raw, int size) throws IOException
+ {
+ input = raw;
+ readBuffer = new byte[size];
+ receiver = new DynamicOutputStream(size, 50);
+ zipOut = new GZIPOutputStream(receiver);
+ }
+
+ public byte[] read() throws IOException
+ {
+ byte[] result = receiver.getBuffer();
+ if (result == null)
+ {
+ return null;
+ }
+ else if (result.length > 0)
+ {
+ return result;
+ }
+
+ int n = input.read(readBuffer);
+ while (true)
+ {
+ if (n > 0)
+ {
+ zipOut.write(readBuffer, 0, n);
+ result = receiver.getBuffer();
+ if ((result != null) && (result.length > 0))
+ {
+ break;
+ }
+ n = input.read(readBuffer);
+ }
+ else
+ {
+ zipOut.close();
+ result = receiver.getBuffer();
+ break;
+ }
+ }
+ return result;
+ }
+ }
+
+ public static void main(String[] args) throws HornetQException, IOException
+ {
+ long begin = System.currentTimeMillis();
+/*
+ FileInputStream input = new
FileInputStream("/home/howard/tmp/jbm.log.1");
+ FileOutputStream output = new
FileOutputStream("/home/howard/tmp/output3.zip");
+ GZIPOutputStream zipOut = new GZIPOutputStream(output);
+
+ byte[] buffer = new byte[1024];
+
+ int n = input.read(buffer);
+
+ int counter = 0;
+
+ while (n > 0)
+ {
+ zipOut.write(buffer, 0, n);
+ counter += n;
+ n = input.read(buffer);
+ }
+ zipOut.close();
+
+ System.out.println("----total output: " + counter);
+*/
+
+ FileInputStream input = new
FileInputStream("/home/howard/tmp/jbm.log.1");
+ FileOutputStream output = new
FileOutputStream("/home/howard/tmp/myzip.zip");
+ GZipPipe pipe = new GZipPipe(input, 2048);
+
+ byte[] buffer;
+
+ buffer = pipe.read();
+
+ while (buffer != null)
+ {
+ //System.out.println("buffer size: " + buffer.length);
+ output.write(buffer);
+ buffer = pipe.read();
+ }
+
+ output.close();
+
+/*
+ FileInputStream input = new
FileInputStream("/home/howard/tmp/jbm.log.1");
+ FileOutputStream output = new
FileOutputStream("/home/howard/tmp/output.zip");
+ ExecutorService service = Executors.newCachedThreadPool();
+ InputStream result = GZipUtil.pipeGZip(input, true, service);
+
+ byte[] buffer = new byte[2048];
+ int n = result.read(buffer);
+ System.out.println("got first data");
+
+ while (n > 0)
+ {
+ output.write(buffer);
+ n = result.read(buffer);
+ }
+*/
+ long end = System.currentTimeMillis();
+
+
+ System.out.println("done. time: " + (end - begin));
+ }
+
}
Modified:
branches/Branch_Large_Message_Compression/tests/src/org/hornetq/tests/integration/client/LargeMessageTest.java
===================================================================
---
branches/Branch_Large_Message_Compression/tests/src/org/hornetq/tests/integration/client/LargeMessageTest.java 2010-11-11
02:23:43 UTC (rev 9872)
+++
branches/Branch_Large_Message_Compression/tests/src/org/hornetq/tests/integration/client/LargeMessageTest.java 2010-11-11
12:03:18 UTC (rev 9873)
@@ -2684,6 +2684,75 @@
}
}
+ public void testLargeMessageCompression() throws Exception
+ {
+ final int messageSize = (int)(3.5 * HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE);
+
+ ClientSession session = null;
+
+ try
+ {
+ server = createServer(true, isNetty());
+
+ server.start();
+
+ ClientSessionFactory sf = createFactory(isNetty());
+
+ session = sf.createSession(false, false, false);
+
+ session.createTemporaryQueue(LargeMessageTest.ADDRESS,
LargeMessageTest.ADDRESS);
+
+ ClientProducer producer = session.createProducer(LargeMessageTest.ADDRESS);
+
+ Message clientFile = createLargeClientMessage(session, messageSize, true);
+
+ producer.send(clientFile);
+
+ session.commit();
+
+ session.start();
+
+ ClientConsumer consumer = session.createConsumer(LargeMessageTest.ADDRESS);
+ ClientMessage msg1 = consumer.receive(1000);
+ msg1.acknowledge();
+ session.commit();
+ Assert.assertNotNull(msg1);
+
+ consumer.close();
+
+ try
+ {
+ msg1.getBodyBuffer().readByte();
+ Assert.fail("Exception was expected");
+ }
+ catch (Throwable ignored)
+ {
+ }
+
+ session.close();
+
+ validateNoFilesOnLargeDir();
+ }
+ finally
+ {
+ try
+ {
+ server.stop();
+ }
+ catch (Throwable ignored)
+ {
+ }
+
+ try
+ {
+ session.close();
+ }
+ catch (Throwable ignored)
+ {
+ }
+ }
+ }
+
// Package protected ---------------------------------------------
// Protected -----------------------------------------------------