Fix the problem that it fail when multiple request need same resource at the same time.

pom.xml

@@ -34,8 +34,8 @@
                 <artifactId>maven-compiler-plugin</artifactId>
                 <version>3.0</version>
                 <configuration>
-                    <source>1.7</source>
-                    <target>1.7</target>
+                    <source>1.8</source>
+                    <target>1.8</target>
                 </configuration>
             </plugin>
             <plugin>
@@ -78,6 +78,24 @@
             <version>4.11</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-all</artifactId>
+            <version>1.9.5</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.powermock</groupId>
+            <artifactId>powermock-api-mockito</artifactId>
+            <version>1.5.4</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.powermock</groupId>
+            <artifactId>powermock-module-junit4</artifactId>
+            <version>1.5.4</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-simple</artifactId>

src/main/java/net/unit8/wscl/ClassLoaderEndpoint.java

@@ -1,11 +1,21 @@
 package net.unit8.wscl;
 
-import net.unit8.wscl.dto.ResourceRequest;
-import net.unit8.wscl.dto.ResourceResponse;
-import net.unit8.wscl.handler.ResourceRequestWriteHandler;
-import net.unit8.wscl.handler.ResourceResponseReadHandler;
-import net.unit8.wscl.util.FressianUtils;
-import net.unit8.wscl.util.PropertyUtils;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Map;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+
+import javax.websocket.ClientEndpoint;
+import javax.websocket.Endpoint;
+import javax.websocket.EndpointConfig;
+import javax.websocket.MessageHandler;
+import javax.websocket.OnOpen;
+import javax.websocket.Session;
+
 import org.fressian.FressianReader;
 import org.fressian.FressianWriter;
 import org.fressian.handlers.ILookup;
@@ -15,12 +25,12 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import javax.websocket.*;
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.util.Map;
-import java.util.concurrent.*;
+import net.unit8.wscl.dto.ResourceRequest;
+import net.unit8.wscl.dto.ResourceResponse;
+import net.unit8.wscl.handler.ResourceRequestWriteHandler;
+import net.unit8.wscl.handler.ResourceResponseReadHandler;
+import net.unit8.wscl.util.FressianUtils;
+import net.unit8.wscl.util.PropertyUtils;
 
 /**
  * @author kawasima
@@ -86,8 +96,8 @@ public Map<String, WriteHandler> valAt(Class key) {
 
         logger.debug("fetch class:" + request.getResourceName() + ":" + request.getClassLoaderId());
 
-        final BlockingQueue<ResourceResponse> queue = new ArrayBlockingQueue<>(1);
-        waitingResponses.put(request.getResourceName(), queue);
+        waitingResponses.putIfAbsent(request.getResourceName(), new ArrayBlockingQueue<>(5));
+        final BlockingQueue<ResourceResponse> queue = waitingResponses.get(request.getResourceName());
         try {
             session.getAsyncRemote().sendBinary(ByteBuffer.wrap(baos.toByteArray()));
             ResourceResponse response = queue.poll(PropertyUtils.getLongSystemProperty("wscl.timeout", 5000), TimeUnit.MILLISECONDS);
@@ -98,7 +108,10 @@ public Map<String, WriteHandler> valAt(Class key) {
         } catch(InterruptedException ex) {
             throw new IOException("Interrupted in waiting for request." + request.getResourceName(), ex);
         } finally {
-            waitingResponses.remove(request.getResourceName());
+            if (waitingResponses.get(request.getResourceName()) != null && waitingResponses.get(request.getResourceName()).isEmpty()) {
+                waitingResponses.remove(request.getResourceName());
+            }
+            fw.close();
         }
     }
 

src/test/java/net/unit8/wscl/ClassLoaderEndpointTest.java

@@ -0,0 +1,124 @@
+package net.unit8.wscl;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.fail;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.nio.ByteBuffer;
+import java.util.UUID;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import javax.websocket.RemoteEndpoint;
+import javax.websocket.RemoteEndpoint.Async;
+import javax.websocket.Session;
+
+import org.fressian.FressianWriter;
+import org.fressian.Writer;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import net.unit8.wscl.dto.ResourceRequest;
+import net.unit8.wscl.dto.ResourceResponse;
+
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({FressianWriter.class,ClassLoaderEndpoint.class})
+public class ClassLoaderEndpointTest {
+    private ClassLoaderEndpoint cle;
+    private Session session;
+    private RemoteEndpoint remoteEndpoint;
+    private ResourceRequest resourceRequest;
+    private FressianWriter fressianWriter;
+
+    @Before
+    public void setup() throws Exception{
+        //setup mocks
+        cle = new ClassLoaderEndpoint();
+        session = mock(Session.class);
+        remoteEndpoint = mock(Async.class);
+        doReturn(remoteEndpoint).when(session).getAsyncRemote();
+        ((Async) doReturn(null).when(remoteEndpoint)).sendBinary(any(ByteBuffer.class));
+        resourceRequest = mock(ResourceRequest.class);
+        doReturn(UUID.randomUUID()).when(resourceRequest).getClassLoaderId();
+        doReturn("resouce1").when(resourceRequest).getResourceName();
+        fressianWriter = mock(FressianWriter.class);
+        doReturn(mock(Writer.class)).when(fressianWriter).writeObject(any(ResourceRequest.class));
+        PowerMockito.whenNew(FressianWriter.class).withAnyArguments().thenReturn(fressianWriter );
+    }
+
+
+
+    @Test
+    public void oneRequest() throws Exception {
+        cle.onOpen(session, null);
+        ExecutorService service = Executors.newCachedThreadPool();
+        CompletableFuture<String> asyncResponse = CompletableFuture.supplyAsync(() -> request(cle,resourceRequest), service);
+        Field f = cle.getClass().getDeclaredField("waitingResponses");
+        f.setAccessible(true);
+        ConcurrentHashMap waitingResponse = (ConcurrentHashMap) f.get(cle);
+        Thread.sleep(1000);
+        BlockingQueue<ResourceResponse> queue = (BlockingQueue<ResourceResponse>) waitingResponse.get("resouce1");
+        queue.offer(new ResourceResponse("resoucename1"));
+        assertThat(asyncResponse.get(),is("resoucename1"));
+    }
+
+    @Test
+    public void twoRequestAtTheSameTime() throws Exception {
+        cle.onOpen(session, null);
+        ExecutorService service = Executors.newCachedThreadPool();
+        CompletableFuture<String> asyncResponse1 = CompletableFuture.supplyAsync(() -> request(cle,resourceRequest), service);
+        CompletableFuture<String> asyncResponse2 = CompletableFuture.supplyAsync(() -> request(cle,resourceRequest), service);
+        Field f = cle.getClass().getDeclaredField("waitingResponses");
+        f.setAccessible(true);
+        ConcurrentHashMap waitingResponse = (ConcurrentHashMap) f.get(cle);
+        Thread.sleep(1000);
+        BlockingQueue<ResourceResponse> queue = (BlockingQueue<ResourceResponse>) waitingResponse.get("resouce1");
+        queue.offer(new ResourceResponse("resoucename1"));
+        queue.offer(new ResourceResponse("resoucename1"));
+        assertThat(asyncResponse1.get(),is("resoucename1"));
+        assertThat(asyncResponse2.get(),is("resoucename1"));
+    }
+
+    @Test
+    public void manyRequestAtTheSameTime() throws Exception {
+        cle.onOpen(session, null);
+        ExecutorService service = Executors.newCachedThreadPool();
+        CompletableFuture<String> asyncResponse1 = CompletableFuture.supplyAsync(() -> request(cle,resourceRequest), service);
+        CompletableFuture<String> asyncResponse2 = CompletableFuture.supplyAsync(() -> request(cle,resourceRequest), service);
+        CompletableFuture<String> asyncResponse3 = CompletableFuture.supplyAsync(() -> request(cle,resourceRequest), service);
+        Field f = cle.getClass().getDeclaredField("waitingResponses");
+        f.setAccessible(true);
+        ConcurrentHashMap waitingResponse = (ConcurrentHashMap) f.get(cle);
+        Thread.sleep(1000);
+        BlockingQueue<ResourceResponse> queue = (BlockingQueue<ResourceResponse>) waitingResponse.get("resouce1");
+        queue.offer(new ResourceResponse("resoucename1"));
+        queue.offer(new ResourceResponse("resoucename1"));
+        queue.offer(new ResourceResponse("resoucename1"));
+
+        assertThat(asyncResponse1.get(),is("resoucename1"));
+        assertThat(asyncResponse2.get(),is("resoucename1"));
+        assertThat(asyncResponse3.get(),is("resoucename1"));
+    }
+
+
+    String request(ClassLoaderEndpoint cle, ResourceRequest resourceRequest){
+        try {
+            return cle.request(resourceRequest).getResourceName();
+        } catch (IOException e) {
+            fail();
+        }
+        return null;
+    }
+}