Fix/msg 369 coap compliance followup

src/main/java/io/mapsmessaging/network/protocol/impl/coap/CoapInterfaceManager.java

@@ -32,6 +32,7 @@
 import java.io.IOException;
 import java.net.SocketAddress;
 import java.nio.channels.SelectionKey;
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
 
@@ -86,14 +87,14 @@ public boolean processPacket(Packet packet) throws IOException {
 
   @Override
   public void close() {
-    for (CoapProtocol protocol : currentSessions.values()) {
+    for (CoapProtocol protocol : new ArrayList<>(currentSessions.values())) {
       try {
         protocol.close();
       } catch (IOException e) {
         // Ignore we are closing
       }
-      currentSessions.clear();
     }
+    currentSessions.clear();
   }
 
   @Override

src/main/java/io/mapsmessaging/network/protocol/impl/coap/CoapProtocol.java

@@ -240,7 +240,7 @@ private void setOptions(@NotNull @NonNull MessageEvent messageEvent, Context con
       optionSet.add(eTag);
     }
     if (context.isObserve()) {
-      Observe option = new Observe((int) (context.getObserveId().incrementAndGet() & 0xffff));
+      Observe option = new Observe((int) (context.getObserveId().incrementAndGet() & 0xffffff));
       optionSet.putOption(option);
     }
     List<SchemaResource> schemas = SchemaManager.getInstance().getSchemaByContext(messageEvent.getDestinationName());

src/main/java/io/mapsmessaging/network/protocol/impl/coap/blockwise/BlockReceiveMonitor.java

@@ -40,7 +40,7 @@ public synchronized void complete(String path) {
     blockBasedPackets.remove(path);
   }
 
-  public void scanForIdle() {
+  public synchronized void scanForIdle() {
     long expired = System.currentTimeMillis() - 30_000;
     List<String> expiredList = new ArrayList<>();
     blockBasedPackets.entrySet().stream()

src/main/java/io/mapsmessaging/network/protocol/impl/coap/blockwise/ReceivePacket.java

@@ -22,6 +22,7 @@
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 
 public class ReceivePacket {
@@ -34,11 +35,18 @@ public ReceivePacket(int blockSize){
     this.blockSize = 1 << ( blockSize +4);
   }
 
-  public void add(int index, byte[] block){
-    while(index > blocks.size()){
-      blocks.add(new byte[0]);
+  public boolean add(int index, byte[] block) {
+    if (index < 0 || block == null || block.length > blockSize) {
+      return false;
     }
-    blocks.add(index, block);
+    if (index < blocks.size()) {
+      return Arrays.equals(blocks.get(index), block);
+    }
+    if (index != blocks.size()) {
+      return false;
+    }
+    blocks.add(block);
+    return true;
   }
 
   public byte[] getFull(){

src/main/java/io/mapsmessaging/network/protocol/impl/coap/listeners/GetListener.java

@@ -122,9 +122,11 @@ private BasePacket buildSubscription(String path, BasePacket request, CoapProtoc
       Context subscriptionContext = protocol.getSubscriptionState().create(path, request);
       subscriptionContext.setSubscribedEventManager(session.addSubscription(context));
       return null;
-    } catch (Exception exception) {
+    } catch (InterruptedException exception) {
       Thread.currentThread().interrupt();
       return request.buildAckResponse(Code.INTERNAL_SERVER_ERROR);
+    } catch (Exception exception) {
+      return request.buildAckResponse(Code.INTERNAL_SERVER_ERROR);
     }
   }
 

src/main/java/io/mapsmessaging/network/protocol/impl/coap/listeners/PublishListener.java

@@ -98,21 +98,40 @@ protected Message build(BasePacket request, CoapProtocol protocol){
 
   private boolean handleBlock(BasePacket request, String path, OptionSet optionSet, CoapProtocol protocol) {
     Block block = (Block) optionSet.getOption(BLOCK1);
-    BlockReceiveState blockReceiveState = protocol.getBlockReceiveMonitor().registerOrGet(block, path);
-    blockReceiveState.getReceivePacket().add(block.getNumber(), request.getPayload());
-    if (request.getType().equals(TYPE.CON)) {
-      BasePacket response = request.buildAckResponse(Code.CONTINUE);
-      try {
-        protocol.sendResponse(response);
-      } catch (IOException e) {
+    byte[] payload = request.getPayload();
+    int blockSize = 1 << (block.getSizeEx() + 4);
+    boolean valid = payload != null && payload.length <= blockSize && (!block.isMore() || payload.length == blockSize);
+    BlockReceiveState blockReceiveState = null;
+    if (valid) {
+      blockReceiveState = protocol.getBlockReceiveMonitor().registerOrGet(block, path);
+      valid = blockReceiveState.getReceivePacket().add(block.getNumber(), payload);
+    }
+    if (!valid) {
+      if (request.getType().equals(TYPE.CON)) {
         try {
-          protocol.close();
-        } catch (IOException ex) {
-          // We are closing
+          protocol.sendResponse(request.buildAckResponse(Code.REQUEST_ENTITY_INCOMPLETE));
+        } catch (IOException e) {
+          try {
+            protocol.close();
+          } catch (IOException ex) {
+            // We are closing
+          }
         }
       }
+      return false;
     }
     if (block.isMore()) {
+      if (request.getType().equals(TYPE.CON)) {
+        try {
+          protocol.sendResponse(request.buildAckResponse(Code.CONTINUE));
+        } catch (IOException e) {
+          try {
+            protocol.close();
+          } catch (IOException ex) {
+            // We are closing
+          }
+        }
+      }
       return false;
     }
     protocol.getBlockReceiveMonitor().complete(path);

src/main/java/io/mapsmessaging/network/protocol/impl/coap/packet/Code.java

@@ -68,6 +68,7 @@ public enum Code {
   NOT_FOUND(ERROR, 4, "Not Found"),
   METHOD_NOT_ALLOWED(ERROR, 5, "Method Not Allowed"),
   NOT_ACCEPTABLE(ERROR, 6, "Not Acceptable"),
+  REQUEST_ENTITY_INCOMPLETE(ERROR, 8, "Request Entity Incomplete"),
   PRECONDITION_FAILED(ERROR, 12, "Precondition Failed"),
   REQUEST_ENTITY_TOO_LARGE(ERROR, 13, "Request Entity Too Large"),
   UNSUPPORTED_CONTENT_FORMAT(ERROR, 15, "Unsupported Content-Format"),
@@ -159,6 +160,8 @@ private static Code getError(int val) {
       case 6:
         return NOT_ACCEPTABLE;
 
+      case 8:
+        return REQUEST_ENTITY_INCOMPLETE;
       case 12:
         return PRECONDITION_FAILED;
       case 13:

src/main/java/io/mapsmessaging/network/protocol/impl/coap/packet/PacketFactory.java

@@ -50,6 +50,9 @@ public BasePacket parseFrame(Packet packet) throws IOException {
     }
 
     int rawCode = packet.get(packet.position() + 1) & 0xff;
+    if (rawCode == EMPTY && (tokenLength != 0 || packet.available() != 4)) {
+      throw new IOException("CoAP Empty message must not contain token, options or payload");
+    }
     int codeClass = (rawCode >>> 5) & 0b111;
     int code = rawCode & 0b11111;
     BasePacket basePacket;

src/main/java/io/mapsmessaging/network/protocol/impl/coap/packet/options/Observe.java

@@ -19,6 +19,8 @@
 
 package io.mapsmessaging.network.protocol.impl.coap.packet.options;
 
+import java.io.IOException;
+
 public class Observe extends BinaryOption {
 
   public Observe() {
@@ -30,6 +32,14 @@ public Observe(int val) {
     this.value = val;
   }
 
+  @Override
+  public void update(byte[] data) throws IOException {
+    if (data.length > 3) {
+      throw new IOException("Observe option exceeds three bytes");
+    }
+    super.update(data);
+  }
+
   public boolean register(){
     return getValue() == 0;
   }

src/test/java/io/mapsmessaging/network/protocol/impl/coap/BlockwiseSendTest.java

@@ -22,6 +22,7 @@
 import java.io.IOException;
 import org.eclipse.californium.core.CoapClient;
 import org.eclipse.californium.core.CoapResponse;
+import org.eclipse.californium.core.coap.CoAP;
 import org.eclipse.californium.core.coap.Request;
 import org.eclipse.californium.elements.exception.ConnectorException;
 import org.junit.jupiter.api.Assertions;
@@ -34,28 +35,38 @@ class BlockwiseSendTest extends BaseCoapTest {
   @ValueSource(ints = {0,1,2,3,4,5,6})
   void basicSendTest(int size) throws ConnectorException, IOException {
     CoapClient client = new CoapClient(getUri());
-    byte[] tmp = new byte[1024];
-    for(int x=0;x<tmp.length;x++){
-      tmp[x] = (byte)(32+(x%36));
-    }
-    int blockSize = 1<<(size+4);
-    int id = 0;
-    byte[] buf = new byte[blockSize];
-    for(int x=0;x<tmp.length;x+=blockSize){
-      Request put = Request.newPut();
-      System.arraycopy(tmp, x, buf, 0, buf.length);
-      put.setPayload(buf);
-      put.getOptions().setBlock1(size, (x + blockSize != tmp.length), id);
-      CoapResponse response1 = client.advanced(put);
-      Assertions.assertNotNull(response1);
-      id++;
-    }
+    client.setTimeout(10_000L);
+    try {
+      byte[] payload = new byte[1024];
+      for (int index = 0; index < payload.length; index++) {
+        payload[index] = (byte) (32 + (index % 36));
+      }
+      int blockSize = 1 << (size + 4);
+      int blockNumber = 0;
+      byte[] blockPayload = new byte[blockSize];
+      for (int offset = 0; offset < payload.length; offset += blockSize) {
+        boolean more = offset + blockSize != payload.length;
+        Request put = Request.newPut();
+        System.arraycopy(payload, offset, blockPayload, 0, blockPayload.length);
+        put.setPayload(blockPayload);
+        put.getOptions().setBlock1(size, more, blockNumber);
+        CoapResponse response = client.advanced(put);
+        Assertions.assertNotNull(response, "No response for block " + blockNumber);
+        if (more) {
+          Assertions.assertEquals(CoAP.ResponseCode.CONTINUE, response.getCode(), "Intermediate block " + blockNumber);
+        } else {
+          Assertions.assertTrue(response.getCode() == CoAP.ResponseCode.CREATED || response.getCode() == CoAP.ResponseCode.CHANGED,
+              "Unexpected final PUT response: " + response.getCode());
+        }
+        blockNumber++;
+      }
 
-    CoapResponse getResponse = client.get();
-
-    Assertions.assertNotNull(getResponse); // Should not be null
-    Assertions.assertArrayEquals(tmp, getResponse.getPayload());
-    client.shutdown();
+      CoapResponse response = client.get();
+      Assertions.assertNotNull(response, "GET did not receive the completed blockwise PUT");
+      Assertions.assertArrayEquals(payload, response.getPayload());
+    } finally {
+      client.shutdown();
+    }
   }
 
 }

src/test/java/io/mapsmessaging/network/protocol/impl/coap/CoapExchangeAndBlockwiseComplianceTest.java

@@ -2,12 +2,15 @@
 
 import io.mapsmessaging.network.protocol.impl.coap.blockwise.BlockReceiveMonitor;
 import io.mapsmessaging.network.protocol.impl.coap.blockwise.BlockReceiveState;
+import io.mapsmessaging.network.protocol.impl.coap.blockwise.ReceivePacket;
 import io.mapsmessaging.network.protocol.impl.coap.blockwise.SendPacket;
 import io.mapsmessaging.network.protocol.impl.coap.listeners.EmptyListener;
+import io.mapsmessaging.network.protocol.impl.coap.listeners.PutListener;
 import io.mapsmessaging.network.protocol.impl.coap.packet.BasePacket;
 import io.mapsmessaging.network.protocol.impl.coap.packet.Code;
 import io.mapsmessaging.network.protocol.impl.coap.packet.TYPE;
 import io.mapsmessaging.network.protocol.impl.coap.packet.options.Block;
+import io.mapsmessaging.network.protocol.impl.coap.packet.options.Constants;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
@@ -170,6 +173,64 @@ void completingBlockTransferRemovesReceiveState() {
     assertNotSame(first, second);
   }
 
+  @Test
+  void nonFinalBlockGetsExactlyOneContinueResponse() throws Exception {
+    CoapProtocol protocol = mock(CoapProtocol.class);
+    when(protocol.getBlockReceiveMonitor()).thenReturn(new BlockReceiveMonitor());
+    BasePacket request = new BasePacket(3, TYPE.CON, Code.EMPTY, 1, 100, new byte[0]);
+    request.getOptions().add(new Block(Constants.BLOCK1, 0, true, 0));
+    request.setPayload(new byte[16]);
+
+    new PutListener().handle(request, protocol);
+
+    ArgumentCaptor<BasePacket> response = ArgumentCaptor.forClass(BasePacket.class);
+    verify(protocol, times(1)).sendResponse(response.capture());
+    assertEquals(TYPE.ACK, response.getValue().getType());
+    assertEquals(Code.CONTINUE, response.getValue().getCode());
+  }
+
+  @Test
+  void missingBlockGetsRequestEntityIncompleteResponse() throws Exception {
+    CoapProtocol protocol = mock(CoapProtocol.class);
+    when(protocol.getBlockReceiveMonitor()).thenReturn(new BlockReceiveMonitor());
+    BasePacket request = new BasePacket(3, TYPE.CON, Code.EMPTY, 1, 101, new byte[0]);
+    request.getOptions().add(new Block(Constants.BLOCK1, 2, false, 0));
+    request.setPayload(new byte[]{1});
+
+    new PutListener().handle(request, protocol);
+
+    ArgumentCaptor<BasePacket> response = ArgumentCaptor.forClass(BasePacket.class);
+    verify(protocol, times(1)).sendResponse(response.capture());
+    assertEquals(Code.REQUEST_ENTITY_INCOMPLETE, response.getValue().getCode());
+  }
+
+  @Test
+  void retransmittedBlockDoesNotDuplicateRepresentation() {
+    ReceivePacket upload = new ReceivePacket(0);
+    byte[] first = sequence(16);
+    byte[] last = new byte[]{42};
+
+    assertTrue(upload.add(0, first));
+    assertTrue(upload.add(0, first));
+    assertTrue(upload.add(1, last));
+    byte[] expected = new byte[first.length + last.length];
+    System.arraycopy(first, 0, expected, 0, first.length);
+    expected[first.length] = last[0];
+    assertArrayEquals(expected, upload.getFull());
+  }
+
+  @Test
+  void incompleteAndConflictingBlocksAreRejectedWithoutCorruptingUpload() {
+    ReceivePacket upload = new ReceivePacket(0);
+    byte[] first = sequence(16);
+
+    assertFalse(upload.add(1, first), "Missing block zero must not become a zero-length placeholder");
+    assertTrue(upload.add(0, first));
+    assertFalse(upload.add(0, new byte[]{42}), "Conflicting retransmission must not insert a duplicate");
+    assertFalse(upload.add(2, new byte[]{42}), "Block gap must be rejected");
+    assertArrayEquals(first, upload.getFull());
+  }
+
   @ParameterizedTest
   @MethodSource("sendPacketShapes")
   void blockPacketSplitsPayloadWithoutLossOrDuplication(int payloadSize, int blockSize) {