You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by ra...@apache.org on 2013/07/19 21:41:25 UTC
svn commit: r1504979 - in
/activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire:
Ack.java Buffer.java BufferInputStream.java ByteBool.java Control.java
Election.java Nack.java PacketType.java StateKey.java package.html
Author: rajdavies
Date: Fri Jul 19 19:41:24 2013
New Revision: 1504979
URL: http://svn.apache.org/r1504979
Log:
Updated to not use protobuf (for now) and use nio2
Added:
activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/Ack.java (with props)
activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/Buffer.java (with props)
activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/BufferInputStream.java (with props)
activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/ByteBool.java (with props)
activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/Control.java (with props)
activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/Election.java (with props)
activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/Nack.java (with props)
activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/PacketType.java (with props)
activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/StateKey.java (with props)
activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/package.html
Added: activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/Ack.java
URL: http://svn.apache.org/viewvc/activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/Ack.java?rev=1504979&view=auto
==============================================================================
--- activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/Ack.java (added)
+++ activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/Ack.java Fri Jul 19 19:41:24 2013
@@ -0,0 +1,98 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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.apache.activeblaze.wire;
+
+import java.io.IOException;
+
+
+/**
+ * An Ack
+ */
+public class Ack extends Packet {
+
+ private long startSequence;
+ private long endSequence;
+
+ public Ack clone() {
+ Ack result = new Ack();
+ copy(result);
+ return result;
+ }
+
+ protected void copy(Ack copy) {
+ super.copy(copy);
+ copy.startSequence = this.startSequence;
+ copy.endSequence = this.endSequence;
+ }
+
+ /**
+ * Get the packetType
+ *
+ * @return the packetType
+ */
+ public int getPacketType() {
+ return PacketType.ACK.getNumber();
+ }
+
+ /**
+ * Get the startSequence
+ *
+ * @return the startSequence
+ */
+ public long getStartSequence() {
+ return this.startSequence;
+ }
+
+ /**
+ * Set the startSequence
+ *
+ * @param startSequence the startSequence to set
+ */
+ public void setStartSequence(long startSequence) {
+ this.startSequence = startSequence;
+ }
+
+ /**
+ * Get the endSequence
+ *
+ * @return the endSequence
+ */
+ public long getEndSequence() {
+ return this.endSequence;
+ }
+
+ /**
+ * Set the endSequence
+ *
+ * @param endSequence the endSequence to set
+ */
+ public void setEndSequence(long endSequence) {
+ this.endSequence = endSequence;
+ }
+
+ public void read(BufferInputStream in) throws IOException {
+ super.read(in);
+ this.startSequence = in.readLong();
+ this.endSequence = in.readLong();
+ }
+
+ public void write(BufferOutputStream out) throws IOException {
+ super.write(out);
+ out.writeLong(this.startSequence);
+ out.writeLong(this.endSequence);
+ }
+}
Propchange: activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/Ack.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/Buffer.java
URL: http://svn.apache.org/viewvc/activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/Buffer.java?rev=1504979&view=auto
==============================================================================
--- activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/Buffer.java (added)
+++ activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/Buffer.java Fri Jul 19 19:41:24 2013
@@ -0,0 +1,202 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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.apache.activeblaze.wire;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+
+public class Buffer implements Comparable<Buffer> {
+ final public byte[] data;
+ final public int offset;
+ final public int length;
+
+ public Buffer(Buffer other) {
+ this(other.data, other.offset, other.length);
+ }
+
+ public Buffer(byte data[]) {
+ this(data, 0, data.length);
+ }
+
+ public Buffer(byte data[], int offset, int length) {
+ this.data = data;
+ this.offset = offset;
+ this.length = length;
+ }
+
+ public Buffer(ByteBuffer byteBuffer) {
+ this.data = new byte[byteBuffer.limit()];
+ byteBuffer.get(data);
+ this.offset = 0;
+ this.length = data.length;
+ }
+
+ public Buffer(String value) {
+ this(value.getBytes());
+ }
+
+ public final Buffer slice(int low, int high) {
+ int sz;
+ if (high < 0) {
+ sz = length + high;
+ } else {
+ sz = high - low;
+ }
+ if (sz < 0) {
+ sz = 0;
+ }
+ return new Buffer(data, offset + low, sz);
+ }
+
+ public final byte[] getData() {
+ return data;
+ }
+
+ public final int getLength() {
+ return length;
+ }
+
+ public final int getOffset() {
+ return offset;
+ }
+
+ public Buffer compact() {
+ if (length != data.length) {
+ return new Buffer(toByteArray());
+ }
+ return this;
+ }
+
+ final public byte[] toByteArray() {
+ if (length != data.length) {
+ byte t[] = new byte[length];
+ System.arraycopy(data, offset, t, 0, length);
+ }
+ return data;
+ }
+
+ public final byte byteAt(int i) {
+ return data[offset + i];
+ }
+
+ @Override
+ public int hashCode() {
+ byte[] target = new byte[4];
+ for (int i = 0; i < length; i++) {
+ target[i % 4] ^= data[offset + i];
+ }
+ return target[0] << 24 | target[1] << 16 | target[2] << 8 | target[3];
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj == this)
+ return true;
+ if (obj == null || obj.getClass() != Buffer.class)
+ return false;
+ return equals((Buffer) obj);
+ }
+
+ final public boolean equals(Buffer obj) {
+ if (length != obj.length) {
+ return false;
+ }
+ for (int i = 0; i < length; i++) {
+ if (obj.data[obj.offset + i] != data[offset + i]) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ final public boolean isEmpty() {
+ return length == 0;
+ }
+
+ final public boolean contains(byte value) {
+ return indexOf(value, 0) >= 0;
+ }
+
+ final public int indexOf(byte value, int pos) {
+ for (int i = pos; i < length; i++) {
+ if (data[offset + i] == value) {
+ return i;
+ }
+ }
+ return -1;
+ }
+
+ final public static Buffer join(List<Buffer> items, Buffer seperator) {
+ if (items.isEmpty())
+ return new Buffer(seperator.data, 0, 0);
+ int size = 0;
+ for (Buffer item : items) {
+ size += item.length;
+ }
+ size += seperator.length * (items.size() - 1);
+ int pos = 0;
+ byte data[] = new byte[size];
+ for (Buffer item : items) {
+ if (pos != 0) {
+ System.arraycopy(seperator.data, seperator.offset, data, pos, seperator.length);
+ pos += seperator.length;
+ }
+ System.arraycopy(item.data, item.offset, data, pos, item.length);
+ pos += item.length;
+ }
+ return new Buffer(data, 0, size);
+ }
+
+ public String toString() {
+ return new String(getData(), getOffset(), getLength());
+ }
+
+ /**
+ * @return a negative integer, zero, or a positive integer as this object is
+ * less than, equal to, or greater than the other buffer.
+ */
+ public int compareTo(Buffer other) {
+ int len1 = this.length;
+ int len2 = other.length;
+ int n = Math.min(len1, len2);
+ byte v1[] = this.data;
+ byte v2[] = other.data;
+ int i = this.offset;
+ int j = other.offset;
+ if (i == j) {
+ int k = i;
+ int lim = n + i;
+ while (k < lim) {
+ byte c1 = v1[k];
+ byte c2 = v2[k];
+ if (c1 != c2) {
+ return c1 - c2;
+ }
+ k++;
+ }
+ } else {
+ while (n-- != 0) {
+ byte c1 = v1[i++];
+ byte c2 = v2[j++];
+ if (c1 != c2) {
+ return c1 - c2;
+ }
+ }
+ }
+ return len1 - len2;
+ }
+}
Propchange: activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/Buffer.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/BufferInputStream.java
URL: http://svn.apache.org/viewvc/activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/BufferInputStream.java?rev=1504979&view=auto
==============================================================================
--- activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/BufferInputStream.java (added)
+++ activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/BufferInputStream.java Fri Jul 19 19:41:24 2013
@@ -0,0 +1,220 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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.apache.activeblaze.wire;
+
+import java.io.DataInput;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * Very similar to the java.io.ByteArrayInputStream but this version is not
+ * thread safe.
+ */
+final public class BufferInputStream extends InputStream implements DataInput {
+ private byte readBuffer[] = new byte[8];
+ private char[] charArray = new char[32];
+ byte buffer[];
+ int limit;
+ int pos;
+ int mark;
+
+ public BufferInputStream(byte data[]) {
+ this(data, 0, data.length);
+ }
+
+ public BufferInputStream(Buffer sequence) {
+ this(sequence.getData(), sequence.getOffset(), sequence.getLength());
+ }
+
+ public BufferInputStream(byte data[], int offset, int size) {
+ this.buffer = data;
+ this.mark = offset;
+ this.pos = offset;
+ this.limit = offset + size;
+ }
+
+ public final int read() {
+ return this.buffer[this.pos++] & 0xff;
+ }
+
+ public int read(byte[] b) {
+ return read(b, 0, b.length);
+ }
+
+ public int read(byte b[], int off, int len) {
+ if (pos < limit) {
+ len = Math.min(len, limit - pos);
+ System.arraycopy(buffer, pos, b, off, len);
+ pos += len;
+ return len;
+ } else {
+ return -1;
+ }
+ }
+
+ public Buffer readBuffer(int len) {
+ Buffer rc = null;
+ if (pos < limit) {
+ len = Math.min(len, limit - pos);
+ rc = new Buffer(buffer, pos, len);
+ pos += len;
+ }
+ return rc;
+ }
+
+ public long skip(long len) {
+ if (pos < limit) {
+ len = Math.min(len, limit - pos);
+ if (len > 0) {
+ pos += len;
+ }
+ return len;
+ } else {
+ return -1;
+ }
+ }
+
+ public int available() {
+ return limit - pos;
+ }
+
+ public boolean markSupported() {
+ return true;
+ }
+
+ public void mark(int markpos) {
+ mark = pos;
+ }
+
+ public void reset() {
+ pos = mark;
+ }
+
+ public boolean readBoolean() {
+ int ch = (this.buffer[this.pos++] & 0xff);
+ return ch != -1 && ch != 0;
+ }
+
+ public byte readByte() {
+ return (byte) (this.buffer[this.pos++] & 0xff);
+ }
+
+ public char readChar() {
+ int ch1 = (this.buffer[this.pos++] & 0xff);
+ int ch2 = (this.buffer[this.pos++] & 0xff);
+ return (char) ((ch1 << 8) + (ch2 << 0));
+ }
+
+ public double readDouble() throws IOException {
+ return Double.longBitsToDouble(readLong());
+ }
+
+ public float readFloat() throws IOException {
+ return Float.intBitsToFloat(readInt());
+ }
+
+ public void readFully(byte[] b) {
+ read(b);
+ }
+
+ public void readFully(byte[] b, int off, int len) {
+ read(b, off, len);
+ }
+
+ public Buffer readBuffer() {
+ Buffer result = null;
+ int len = readInt();
+ if (len > 0) {
+ byte[] data = new byte[len];
+ readFully(data);
+ result = new Buffer(data);
+
+ }
+ return result;
+ }
+
+ public int readInt() {
+ int ch1 = (this.buffer[this.pos++] & 0xff);
+ int ch2 = (this.buffer[this.pos++] & 0xff);
+ int ch3 = (this.buffer[this.pos++] & 0xff);
+ int ch4 = (this.buffer[this.pos++] & 0xff);
+ return ((ch1 << 24) + (ch2 << 16) + (ch3 << 8) + (ch4 << 0));
+ }
+
+ public String readLine() {
+ return null;
+ }
+
+ public long readLong() {
+ read(this.readBuffer, 0, 8);
+ return (((long) this.readBuffer[0] << 56) + ((long) (this.readBuffer[1] & 255) << 48)
+ + ((long) (this.readBuffer[2] & 255) << 40) + ((long) (this.readBuffer[3] & 255) << 32)
+ + ((long) (this.readBuffer[4] & 255) << 24) + ((this.readBuffer[5] & 255) << 16)
+ + ((this.readBuffer[6] & 255) << 8) + ((this.readBuffer[7] & 255) << 0));
+ }
+
+ public short readShort() {
+ int ch1 = (this.buffer[this.pos++] & 0xff);
+ int ch2 = (this.buffer[this.pos++] & 0xff);
+ return (short) ((ch1 << 8) + (ch2 << 0));
+ }
+
+ public String readUTF() throws IOException {
+ return DataInputStream.readUTF(this);
+ }
+
+ public String readString() {
+ String result = null;
+ int len = readShort();
+ if (len > 0) {
+ if (len > this.charArray.length) {
+ this.charArray = new char[len];
+ }
+ for (int i = 0; i < len; i++) {
+ this.charArray[i] = (char) (this.buffer[this.pos++] & 0xff);
+ }
+ result = String.copyValueOf(this.charArray, 0, len);
+ } else if (len == 0) {
+ result = new String();
+ }
+ return result;
+ }
+
+ public Object readObject() throws IOException {
+ return IOUtils.readObject(this);
+ }
+
+ public int readUnsignedByte() {
+ return (this.buffer[this.pos++] & 0xff);
+ }
+
+ public int readUnsignedShort() {
+ int ch1 = (this.buffer[this.pos++] & 0xff);
+ int ch2 = (this.buffer[this.pos++] & 0xff);
+ return (ch1 << 8) + (ch2 << 0);
+ }
+
+ public int skipBytes(int n) {
+ int total = 0;
+ int cur = 0;
+ while ((total < n) && ((cur = (int) skip(n - total)) > 0)) {
+ total += cur;
+ }
+ return total;
+ }
+}
Propchange: activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/BufferInputStream.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/ByteBool.java
URL: http://svn.apache.org/viewvc/activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/ByteBool.java?rev=1504979&view=auto
==============================================================================
--- activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/ByteBool.java (added)
+++ activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/ByteBool.java Fri Jul 19 19:41:24 2013
@@ -0,0 +1,78 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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.apache.activeblaze.wire;
+
+/**
+ * Store booleans in a byte
+ */
+public final class ByteBool {
+ private byte data;
+ private int pos = 0;
+
+ /**
+ * write a boolean value - and move the position in the byte 1 bit forward
+ */
+ public void writeBoolean(boolean value) {
+ if (this.pos >= 8) {
+ throw new RuntimeException("ByteBoolean overwrite: " + this.pos);
+ }
+ if (value) {
+ this.data |= 0x01 << this.pos;
+ }
+ this.pos++;
+ }
+
+ /**
+ * read a boolean value - and move the position in the byte 1 bit forward
+ *
+ * @return true if the bit is set
+ */
+ public boolean readBoolean() {
+ boolean result = ((this.data >> this.pos) & 0x01) != 0;
+ this.pos++;
+ return result;
+ }
+
+ /**
+ * @return the underlying byte
+ */
+ public byte getData() {
+ return this.data;
+ }
+
+ /**
+ * Set the byte to read
+ */
+ public void setData(byte v) {
+ this.data = v;
+ }
+
+ /**
+ * reset the position and clear the byte
+ */
+ public void clear() {
+ this.data = 0;
+ this.pos = 0;
+ }
+
+ /**
+ * reset the position to the first bit
+ */
+ public void reset() {
+ this.pos = 0;
+ }
+}
Propchange: activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/ByteBool.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/Control.java
URL: http://svn.apache.org/viewvc/activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/Control.java?rev=1504979&view=auto
==============================================================================
--- activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/Control.java (added)
+++ activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/Control.java Fri Jul 19 19:41:24 2013
@@ -0,0 +1,74 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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.apache.activeblaze.wire;
+
+import java.io.IOException;
+
+/**
+ * A Control
+ */
+public class Control extends Packet {
+ private long lastMessageSequence;
+
+ public Control clone() {
+ Control result = new Control();
+ copy(result);
+ return result;
+ }
+
+ protected void copy(Control copy) {
+ super.copy(copy);
+ copy.lastMessageSequence = this.lastMessageSequence;
+ }
+
+ /**
+ * Get the packetType
+ *
+ * @return the packetType
+ */
+ public int getPacketType() {
+ return PacketType.CONTROL.getNumber();
+ }
+
+ /**
+ * Get the lastMessageSequence
+ *
+ * @return the lastMessageSequence
+ */
+ public long getLastMessageSequence() {
+ return this.lastMessageSequence;
+ }
+
+ /**
+ * Set the lastMessageSequence
+ *
+ * @param lastMessageSequence the lastMessageSequence to set
+ */
+ public void setLastMessageSequence(long lastMessageSequence) {
+ this.lastMessageSequence = lastMessageSequence;
+ }
+
+ public void read(BufferInputStream in) throws IOException {
+ super.read(in);
+ this.lastMessageSequence = in.readLong();
+ }
+
+ public void write(BufferOutputStream out) throws IOException {
+ super.write(out);
+ out.writeLong(this.lastMessageSequence);
+ }
+}
Propchange: activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/Control.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/Election.java
URL: http://svn.apache.org/viewvc/activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/Election.java?rev=1504979&view=auto
==============================================================================
--- activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/Election.java (added)
+++ activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/Election.java Fri Jul 19 19:41:24 2013
@@ -0,0 +1,94 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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.apache.activeblaze.wire;
+
+import java.io.IOException;
+
+/**
+ * An Election
+ */
+public class Election extends Packet {
+ public enum ElectionType {ELECTION, ANSWER, MASTER}
+
+ ;
+ private String memberId;
+ private ElectionType electionType;
+
+ public int getPacketType() {
+ return PacketType.ELECTION.getNumber();
+ }
+
+ public Election clone() {
+ Election result = new Election();
+ copy(result);
+ return result;
+ }
+
+ protected void copy(Election copy) {
+ super.copy(copy);
+ copy.memberId = this.memberId;
+ copy.electionType = this.electionType;
+ }
+
+ /**
+ * Get the memberId
+ *
+ * @return the memberId
+ */
+ public String getMemberId() {
+ return this.memberId;
+ }
+
+ /**
+ * Set the memberId
+ *
+ * @param memberId the memberId to set
+ */
+ public void setMemberId(String memberId) {
+ this.memberId = memberId;
+ }
+
+ /**
+ * Get the electionType
+ *
+ * @return the electionType
+ */
+ public ElectionType getElectionType() {
+ return this.electionType;
+ }
+
+ /**
+ * Set the electionType
+ *
+ * @param electionType the electionType to set
+ */
+ public void setElectionType(ElectionType electionType) {
+ this.electionType = electionType;
+ }
+
+ public void read(BufferInputStream in) throws IOException {
+ super.read(in);
+ this.memberId = in.readString();
+ this.electionType = ElectionType.valueOf(in.readString());
+ }
+
+ public void write(BufferOutputStream out) throws IOException {
+ super.write(out);
+ out.writeString(this.memberId);
+ out.writeString(this.electionType.toString());
+ }
+}
Propchange: activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/Election.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/Nack.java
URL: http://svn.apache.org/viewvc/activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/Nack.java?rev=1504979&view=auto
==============================================================================
--- activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/Nack.java (added)
+++ activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/Nack.java Fri Jul 19 19:41:24 2013
@@ -0,0 +1,37 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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.apache.activeblaze.wire;
+
+/**
+ * A Nack
+ */
+public class Nack extends Ack {
+ /**
+ * Get the packetType
+ *
+ * @return the packetType
+ */
+ public int getPacketType() {
+ return PacketType.NACK.getNumber();
+ }
+
+ public Nack clone() {
+ Nack result = new Nack();
+ copy(result);
+ return result;
+ }
+}
Propchange: activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/Nack.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/PacketType.java
URL: http://svn.apache.org/viewvc/activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/PacketType.java?rev=1504979&view=auto
==============================================================================
--- activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/PacketType.java (added)
+++ activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/PacketType.java Fri Jul 19 19:41:24 2013
@@ -0,0 +1,136 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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.apache.activeblaze.wire;
+
+import org.apache.activeblaze.BlazeMessage;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+/**
+ * The Type of Packet
+ */
+public enum PacketType {
+
+ /**
+ *
+ */
+ PACKET("PACKET", 0),
+ MEMBER("MEMBER", 1),
+ ELECTION("ELECTION", 2),
+ ACK("ACK", 3),
+ NACK("NACK", 4),
+ STATE("STATE", 5),
+ CONTROL("CONTROL", 6),
+ MESSAGE("MESSAGE", 7),
+ JMS_MESSAGE("JMS_MESSAGE", 8),
+ JMS_BYTES("JMS_BYTES", 9),
+ JMS_MAP("JMS_MAP", 10),
+ JMS_OBJECT("JMS_OBJECT", 11),
+ JMS_STREAM("JMS_STREAM", 12),
+ JMS_TEXT("JMS_TEXT", 13);
+ private static final Log LOG = LogFactory.getLog(PacketType.class);
+ private final String name;
+ private final int value;
+
+ private PacketType(String name, int value) {
+ this.name = name;
+ this.value = value;
+ }
+
+ public final int getNumber() {
+ return value;
+ }
+
+ public final String toString() {
+ return name;
+ }
+
+ public static PacketType valueOf(int value) {
+ switch (value) {
+ case 0:
+ return PACKET;
+ case 1:
+ return MEMBER;
+ case 2:
+ return ELECTION;
+ case 3:
+ return ACK;
+ case 4:
+ return NACK;
+ case 5:
+ return STATE;
+ case 6:
+ return CONTROL;
+ case 7:
+ return MESSAGE;
+ case 8:
+ return JMS_MESSAGE;
+ case 9:
+ return JMS_BYTES;
+ case 10:
+ return JMS_MAP;
+ case 11:
+ return JMS_OBJECT;
+ case 12:
+ return JMS_STREAM;
+ case 13:
+ return JMS_TEXT;
+ default:
+ LOG.error("Unknown type" + value);
+ return null;
+ }
+ }
+
+
+ public Packet createPacket() {
+ switch (this) {
+ case PACKET:
+ return new Packet();
+ case MEMBER:
+ return new MemberImpl();
+ case ELECTION:
+ return new Election();
+ case ACK:
+ return new Ack();
+ case NACK:
+ return new Nack();
+ case STATE:
+ return new StateValue();
+ case CONTROL:
+ return new Control();
+ case MESSAGE:
+ return new BlazeMessage();
+ case JMS_MESSAGE:
+ return new BlazeJmsMessage();
+ case JMS_BYTES:
+ return new BlazeJmsBytesMessage();
+ case JMS_MAP:
+ return new BlazeJmsMapMessage();
+ case JMS_OBJECT:
+ return new BlazeJmsObjectMessage();
+ case JMS_STREAM:
+ return new BlazeJmsStreamMessage();
+ case JMS_TEXT:
+ return new BlazeJmsTextMessage();
+
+ default:
+ return null;
+ }
+ }
+
+
+}
Propchange: activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/PacketType.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/StateKey.java
URL: http://svn.apache.org/viewvc/activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/StateKey.java?rev=1504979&view=auto
==============================================================================
--- activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/StateKey.java (added)
+++ activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/StateKey.java Fri Jul 19 19:41:24 2013
@@ -0,0 +1,274 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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.apache.activeblaze.wire;
+
+import java.io.IOException;
+
+/**
+ * Holds information about a StateKey
+ */
+public class StateKey implements Cloneable {
+ private MemberImpl owner;
+ private String key;
+ private boolean locked;
+ private boolean removeOnExit;
+ private boolean releaseLockOnExit;
+ private long expiration;
+ private long lockExpiration;
+
+ /**
+ * Constructor
+ */
+ public StateKey(MemberImpl owner, String key) {
+ this.owner = owner;
+ this.key = key;
+ }
+
+ public StateKey() {
+ }
+
+ public StateKey clone() {
+ StateKey result = new StateKey();
+ copy(result);
+ return result;
+ }
+
+ protected void copy(StateKey copy) {
+ copy.owner = this.owner.clone();
+ copy.key = this.key;
+ copy.locked = this.locked;
+ copy.removeOnExit = this.removeOnExit;
+ copy.releaseLockOnExit = this.releaseLockOnExit;
+ copy.expiration = this.expiration;
+ copy.lockExpiration = this.lockExpiration;
+ }
+
+ /**
+ * @return the owner
+ */
+ public MemberImpl getOwner() {
+ return this.owner;
+ }
+
+ /**
+ * @param member
+ */
+ public void setOwner(MemberImpl member) {
+ this.owner = member;
+ }
+
+ /**
+ * Get the key
+ *
+ * @return the key
+ */
+ public String getKey() {
+ return this.key;
+ }
+
+ /**
+ * Set the key
+ *
+ * @param key the key to set
+ */
+ public void setKey(String key) {
+ this.key = key;
+ }
+
+ /**
+ * Get the locked
+ *
+ * @return the locked
+ */
+ public boolean isLocked() {
+ return this.locked;
+ }
+
+ /**
+ * Set the locked
+ *
+ * @param locked the locked to set
+ */
+ public void setLocked(boolean locked) {
+ this.locked = locked;
+ }
+
+ /**
+ * Get the removeOnExit
+ *
+ * @return the removeOnExit
+ */
+ public boolean isRemoveOnExit() {
+ return this.removeOnExit;
+ }
+
+ /**
+ * Set the removeOnExit
+ *
+ * @param removeOnExit the removeOnExit to set
+ */
+ public void setRemoveOnExit(boolean removeOnExit) {
+ this.removeOnExit = removeOnExit;
+ }
+
+ /**
+ * Get the releaseLockOnExit
+ *
+ * @return the releaseLockOnExit
+ */
+ public boolean isReleaseLockOnExit() {
+ return this.releaseLockOnExit;
+ }
+
+ /**
+ * Set the releaseLockOnExit
+ *
+ * @param releaseLockOnExit the releaseLockOnExit to set
+ */
+ public void setReleaseLockOnExit(boolean releaseLockOnExit) {
+ this.releaseLockOnExit = releaseLockOnExit;
+ }
+
+ /**
+ * Get the expiration
+ *
+ * @return the expiration
+ */
+ public long getExpiration() {
+ return this.expiration;
+ }
+
+ /**
+ * Set the expiration
+ *
+ * @param expiration the expiration to set
+ */
+ public void setExpiration(long expiration) {
+ this.expiration = expiration;
+ }
+
+ /**
+ * Get the lockExpiration
+ *
+ * @return the lockExpiration
+ */
+ public long getLockExpiration() {
+ return this.lockExpiration;
+ }
+
+ /**
+ * Set the lockExpiration
+ *
+ * @param lockExpiration the lockExpiration to set
+ */
+ public void setLockExpiration(long lockExpiration) {
+ this.lockExpiration = lockExpiration;
+ }
+
+ /**
+ * set time to live
+ */
+ public void setTimeToLive(long ttl) {
+ if (ttl > 0) {
+ setExpiration(ttl + System.currentTimeMillis());
+ } else {
+ setExpiration(0);
+ }
+ }
+
+ /**
+ * set lease time to live
+ */
+ public void setLockLeaseTime(long ttl) {
+ if (ttl > 0) {
+ setLockExpiration(ttl + System.currentTimeMillis());
+ } else {
+ setLockExpiration(0);
+ }
+ }
+
+ /**
+ * @return true if expired
+ */
+ public boolean isExpired() {
+ return isExpired(System.currentTimeMillis());
+ }
+
+ /**
+ * @return true if expired
+ */
+ public boolean isExpired(long currentTime) {
+ boolean result = getExpiration() > 0 && getExpiration() < currentTime;
+ return result;
+ }
+
+ /**
+ * @return true if lock expired
+ */
+ public boolean isLockExpired() {
+ return isLockExpired(System.currentTimeMillis());
+ }
+
+ /**
+ * @return true if lock expired
+ */
+ public boolean isLockExpired(long currentTime) {
+ return getLockExpiration() > 0 && getLockExpiration() < currentTime;
+ }
+
+ public int hashCode() {
+ return getKey() != null ? getKey().hashCode() : super.hashCode();
+ }
+
+ public boolean equals(Object obj) {
+ boolean result = false;
+ if (obj instanceof StateKey) {
+ StateKey other = (StateKey) obj;
+ result = other.getKey().equals(getKey());
+ }
+ return result;
+ }
+
+ public String toString() {
+ return "key:" + getKey();
+ }
+
+ public void write(BufferOutputStream out) throws IOException {
+ this.owner.write(out);
+ out.writeUTF(this.key);
+ ByteBool byteBool = new ByteBool();
+ byteBool.writeBoolean(this.locked);
+ byteBool.writeBoolean(this.removeOnExit);
+ byteBool.writeBoolean(this.releaseLockOnExit);
+ out.writeByte(byteBool.getData());
+ out.writeLong(this.expiration);
+ out.writeLong(this.lockExpiration);
+ }
+
+ public void read(BufferInputStream in) throws IOException {
+ this.owner = new MemberImpl();
+ this.owner.read(in);
+ this.key = in.readUTF();
+ ByteBool byteBool = new ByteBool();
+ byteBool.setData(in.readByte());
+ this.locked = byteBool.readBoolean();
+ this.removeOnExit = byteBool.readBoolean();
+ this.releaseLockOnExit = byteBool.readBoolean();
+ this.expiration = in.readLong();
+ this.lockExpiration = in.readLong();
+ }
+}
Propchange: activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/StateKey.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/package.html
URL: http://svn.apache.org/viewvc/activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/package.html?rev=1504979&view=auto
==============================================================================
--- activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/package.html (added)
+++ activemq/activemq-blaze/trunk/src/main/java/org/apache/activeblaze/wire/package.html Fri Jul 19 19:41:24 2013
@@ -0,0 +1,25 @@
+!--
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF 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.
+-->
+<html>
+<head>
+</head>
+<body>
+
+Command types
+
+</body>
+</html>
\ No newline at end of file