Headers scott

src/main/java/io/nats/client/Message.java

@@ -13,6 +13,8 @@
 
 package io.nats.client;
 
+import io.nats.client.impl.Headers;
+
 /**
  * The NATS library uses a Message object to encapsulate incoming messages. Applications
  * publish and send requests with raw strings and byte[] but incoming messages can have a few
@@ -22,35 +24,56 @@
  * and is safe to manipulate.
  */
 public interface Message {
+	byte[] EMPTY_BODY = new byte[0];
 
 	/**
 	 * @return the subject that this message was sent to
 	 */
-	public String getSubject();
+	String getSubject();
 
 	/**
 	 * @return the subject the application is expected to send a reply message on
 	 */
-	public String getReplyTo();
+	String getReplyTo();
+
+	/**
+	 * @return the headers from the message
+	 */
+	Headers getHeaders();
+
+	/**
+	 * @return if is utf8Mode
+	 */
+	boolean isUtf8mode();
 
 	/**
 	 * @return the data from the message
 	 */
-	public byte[] getData();
+	byte[] getData();
 
 	/**
 	 * @return the Subscription associated with this message, may be owned by a Dispatcher
 	 */
-	public Subscription getSubscription();
+	Subscription getSubscription();
 
 	/**
 	 * @return the id associated with the subscription, used by the connection when processing an incoming
 	 * message from the server
 	 */
-	public String getSID();
+	String getSID();
 
 	/**
 	 * @return the connection which can be used for publishing, will be null if the subscription is null
 	 */
-	public Connection getConnection();
+	Connection getConnection();
+
+	/**
+	 * @return the protocol bytes
+	 */
+	byte[] getProtocolBytes();
+
+	/**
+	 * @return the message size in bytes
+	 */
+	long getSizeInBytes();
 }

src/main/java/io/nats/client/Nats.java

@@ -13,10 +13,10 @@
 
 package io.nats.client;
 
-import java.io.IOException;
-
 import io.nats.client.impl.NatsImpl;
 
+import java.io.IOException;
+
 /**
  * The Nats class is the entry point into the NATS client for Java. This class
  * is used to create a connection to the NATS server. Connecting is a

src/main/java/io/nats/client/impl/ByteBufferUtil.java

@@ -0,0 +1,15 @@
+package io.nats.client.impl;
+
+import java.nio.ByteBuffer;
+
+public class ByteBufferUtil {
+
+    public static ByteBuffer enlargeBuffer(ByteBuffer buffer, int atLeast) {
+        int current = buffer.capacity();
+        int newSize = Math.max(current * 2, atLeast);
+        ByteBuffer newBuffer = ByteBuffer.allocate(newSize);
+        buffer.flip();
+        newBuffer.put(buffer);
+        return newBuffer;
+    }
+}

src/main/java/io/nats/client/impl/Headers.java

@@ -0,0 +1,188 @@
+// Copyright 2015-2018 The NATS Authors
+// Licensed 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 io.nats.client.impl;
+
+import java.util.*;
+
+public class Headers {
+	private static final String KEY_CANNOT_BE_EMPTY_OR_NULL = "Header key cannot be null.";
+	private static final String VALUES_CANNOT_BE_EMPTY_OR_NULL = "Header values cannot be empty or null.";
+
+	private final Map<String, Set<String>> headerMap = new HashMap<>();
+
+	/**
+	 * If the key is present add the values to the set of values for the key.
+	 * If the key is not present, sets the specified values for the key.
+	 * Duplicate values are ignored. Null and empty values are not allowed
+	 *
+	 * @param key the key
+	 * @param values the values
+	 * @return {@code true} if this object did not already contain values for the key
+	 * @throws IllegalArgumentException if the key is null or empty
+	 *         -or- any value is null or empty.
+	 */
+	public boolean add(String key, String... values) {
+		return add(key, Arrays.asList(values));
+	}
+
+	/**
+	 * If the key is present add the values to the set of values for the key.
+	 * If the key is not present, sets the specified values for the key.
+	 * Duplicate values are ignored. Null and empty values are not allowed.
+	 *
+	 * @param key the key
+	 * @param values the values
+	 * @return {@code true} if this object did not already contain the key or the
+	 *         values for the key changed.
+	 * @throws IllegalArgumentException if the key is null or empty
+	 *         -or- if then input collection is null
+	 *         -or- if any item in the collection is null or empty.
+	 */
+	public boolean add(String key, Collection<String> values) {
+		Set<String> validatedSet = validateKeyAndValues(key, values);
+
+		Set<String> currentSet = headerMap.get(key);
+		if (currentSet == null) {
+			headerMap.put(key, validatedSet);
+			return true;
+		}
+
+		return currentSet.addAll(validatedSet);
+	}
+
+	private Set<String> validateKeyAndValues(String key, Collection<String> values) {
+		keyCannotBeNull(key);
+		valuesCannotBeEmptyOrNull(values);
+		Set<String> validatedSet = new HashSet<>();
+		for (String v : values) {
+			valueCannotBeEmptyOrNull(v);
+			validatedSet.add(v);
+		}
+		return validatedSet;
+	}
+
+	/**
+	 * Associates the specified values with the key. If the key was already present
+	 * any existing values are removed and replaced with the new set.
+	 * Duplicate values are ignored. Null and empty values are not allowed
+	 *
+	 * @param key the key
+	 * @param values the values
+	 * @return {@code true} if this object did not already contain values for the key
+	 * @throws IllegalArgumentException if the key is null or empty
+	 *         -or- any value is null or empty.
+	 */
+	public boolean put(String key, String... values) {
+		return put(key, Arrays.asList(values));
+	}
+
+	/**
+	 * Associates the specified values with the key. If the key was already present
+	 * any existing values are removed and replaced with the new set.
+	 * Duplicate values are ignored. Null and empty values are not allowed
+	 *
+	 * @param key the key
+	 * @param values the values
+	 * @return {@code true} if this object did not already contain values for the key
+	 * @throws IllegalArgumentException if the key is null or empty
+	 *         -or- if then input collection is null
+	 *         -or- if any item in the collection is null or empty.
+	 */
+	public boolean put(String key, Collection<String> values) {
+		Set<String> validatedSet = validateKeyAndValues(key, values);
+		return headerMap.put(key, validatedSet) == null;
+	}
+
+	/**
+	 * Removes each key and its values if the key was present
+	 *
+	 * @param keys the key or keys to remove
+	 * @return {@code true} if any key was present
+	 */
+	public boolean remove(String... keys) {
+		return remove(Arrays.asList(keys));
+	}
+
+	/**
+	 * Removes each key and its values if the key was present
+	 *
+	 * @param keys the key or keys to remove
+	 * @return {@code true} if any key was present
+	 */
+	public boolean remove(Collection<String> keys) {
+		boolean changed = false;
+		for (String key : keys) {
+			if (headerMap.remove(key) != null) {
+				changed = true;
+			}
+		}
+		return changed;
+	}
+
+	public int size() {
+		return headerMap.size();
+	}
+
+	public boolean isEmpty() {
+		return headerMap.isEmpty();
+	}
+
+	public void clear() {
+		headerMap.clear();
+	}
+
+	public boolean containsKey(String key) {
+		return headerMap.containsKey(key);
+	}
+
+	public Set<String> values(String key) {
+		Set<String> set = headerMap.get(key);
+		return set == null ? null : Collections.unmodifiableSet(set);
+	}
+
+	public Set<String> keySet() {
+		return headerMap.keySet();
+	}
+
+	private void keyCannotBeNull(String key) {
+		if (key == null || key.length() == 0) {
+			throw new IllegalArgumentException(KEY_CANNOT_BE_EMPTY_OR_NULL);
+		}
+	}
+
+	private void valueCannotBeEmptyOrNull(String val) {
+		if (val == null || val.length() == 0) {
+			throw new IllegalArgumentException(VALUES_CANNOT_BE_EMPTY_OR_NULL);
+		}
+	}
+
+	private void valuesCannotBeEmptyOrNull(Collection<String> vals) {
+		if (vals == null || vals.size() == 0) {
+			throw new IllegalArgumentException(VALUES_CANNOT_BE_EMPTY_OR_NULL);
+		}
+	}
+
+	@Override
+	public boolean equals(Object o) {
+		if (this == o) return true;
+		if (o == null || getClass() != o.getClass()) return false;
+		Headers headers = (Headers) o;
+		return Objects.equals(headerMap, headers.headerMap);
+	}
+
+	@Override
+	public int hashCode() {
+		return Objects.hash(headerMap);
+	}
+}

src/main/java/io/nats/client/impl/MessageQueue.java

@@ -58,7 +58,7 @@ class MessageQueue {
         this.length = new AtomicLong(0);
 
         // The poisonPill is used to stop poll and accumulate when the queue is stopped
-        this.poisonPill = new NatsMessage("_poison", null, NatsConnection.EMPTY_BODY, false);
+        this.poisonPill = new NatsMessage.PublishBuilder().subject("_poison").build();
 
         this.filterLock = new ReentrantLock();
 
@@ -240,9 +240,9 @@ NatsMessage accumulate(long maxSize, long maxMessages, Duration timeout)
                 if (maxSize<0 || (size + s) < maxSize) { // keep going
                     size += s;
                     count++;
-                    
-                    cursor.next = this.queue.poll();
-                    cursor = cursor.next;
+
+                    cursor.setNext(this.queue.poll());
+                    cursor = cursor.getNext();
 
                     if (count == maxMessages) {
                         break;