单元测试Java多线程网络应用程序

Question

我正在编写一个Java多线程网络应用程序，并且真正有困难想出一种方法来单元测试发送和接收来自网络客户端的通信的对象。

该对象向多个客户端发送消息，然后等待来自客户端的响应。

随着每个客户端的响应，更新了仪表板式GUI。

更详细...

发送消息对象表示要被发送的文本消息，并且包含客户端的一个阵列，其应接收该消息。

消息对象负责将自己分配给所有适当的客户端。

当在Message对象上调用dispatch（）方法时，该对象会为客户端阵列中的每个客户端生成一个新线程（MessageDispatcher）。

每个MessageDispatcher：

• 打开一个新的TCP套接字（Socket）连接到客户端

• 将消息传递给它的客户... PrintWriter的通过out.println（MSG文本）

• 创建一个'Status'对象，该对象被传递给Message对象中的Queue，然后传递给GUI。

每个状态对象表示以下事件之一：

• 消息传递到插座（通过为PrintWriter通过out.println（））从客户机接收到

• 显示收据（经由的BufferedReader/InputStreamReader in.readline（）... 阻塞，直到收到网络输入为止）

• 用户确认从客户端收到的收据（通过与上面相同的方法）

所以..我想单元测试Message对象。 （使用JUnit）

单元测试称为MessageTest.java（下面包含）。

我的第一步是用单个收件人设置一个Message对象。

然后我使用JMockit来创建一个模拟Socket对象，它可以向PrintWriter提供一个模拟OutputStream对象（我正在使用扩展OutputStream的ByteArrayOutputStream）。

然后，当MessageDispatcher调用（PrintWriter对象）.out时，消息文本将理想地传递给我的模拟Socket对象（通过模拟OutputStream），它可以检查消息文本是否正确。

而对于使用InputStreamReader样品原理....的模拟Socket对象还提供供给一个模拟的BufferedReader其由MessageDispatcher称为（如前面提到的，在in.readLine MessageDispatcher块（））一个模拟InputStreamReader的对象。此时模拟的BufferedReader应该提供一个假的确认到MessageDispatcher ...

// mock Socket 
Mockit.redefineMethods(Socket.class, new Object() 
{ 

    ByteArrayOutputStream output = new ByteArrayOutputStream(); 
    ByteArrayInputStream input = new ByteArrayInputStream(); 

    public OutputStream getOutputStream() 
    { 
     return output; 
    } 

    public InputStream getInputStream() 
    { 
     return input; 
    } 

}); 

如果这不是多线程的，这应该一切工作正常。不过，我不知道如何用多线程来做到这一点。任何人都可以给我任何建议或提示？

此外，如果您对设计有任何意见（例如，Message对象负责自己的交付，而不是单独的交付对象..“依赖注入”风格/单独的线程为每个客户交付），那么我会感兴趣也听到了。

UPDATE：这里是代码：

Message.java

public class Message { 

    Client[] to; 

    String contents; 

    String status; 

    StatusListener listener; 

    BlockingQueue<Status> statusQ; 

    public Message(Client[] to, String contents, StatusListener listener) 
    { 
     this.to = to; 
     this.contents = contents; 
     this.listener = listener; 
    } 

    public void dispatch() 
    { 
     try { 

      // open a new thread for each client 

      // keep a linked list of socket references so that all threads can be closed 
      List<Socket> sockets = Collections.synchronizedList(new ArrayList<Socket>()); 

      // initialise the statusQ for threads to report message status 
      statusQ = new ArrayBlockingQueue<Status>(to.length*3); // max 3 status objects per thread 

      // dispatch to each client individually and wait for confirmation 
      for (int i=0; i < to.length; i++) { 

      System.out.println("Started new thread"); 

      (new Thread(new MessageDispatcher(to[i], contents, sockets, statusQ))).start(); 

      } 

      // now, monitor queue and empty the queue as it fills up.. (consumer) 
      while (true) { 
       listener.updateStatus(statusQ.take()); 
      } 
     } 

     catch (Exception e) { e.printStackTrace(); } 

    } 

    // one MessageDispatcher per client 
    private class MessageDispatcher implements Runnable 
    { 

     private Client client; 
     private String contents; 
     private List<Socket> sockets; 
     private BlockingQueue<Status> statusQ; 

     public MessageDispatcher(Client client, String contents, List<Socket> sockets, BlockingQueue<Status> statusQ) { 

      this.contents = contents; 

      this.client = client; 

      this.sockets = sockets; 

      this.statusQ = statusQ; 

     } 

     public void run() { 

     try { 

      // open socket to client 
      Socket sk = new Socket(client.getAddress(), CLIENTPORT); 

      // add reference to socket to list 
      synchronized(sockets) { 
       sockets.add(sk); 
      } 

      PrintWriter out = new PrintWriter(sk.getOutputStream(), true); 

      BufferedReader in = new BufferedReader(new InputStreamReader(sk.getInputStream())); 

      // send message 
      out.println(contents); 

      // confirm dispatch 
      statusQ.add(new Status(client, "DISPATCHED")); 

      // wait for display receipt 
      in.readLine(); 

      statusQ.add(new Status(client, "DISPLAYED")); 

      // wait for read receipt 
      in.readLine(); 

      statusQ.add(new Status(client, "READ")); 

      } 

      catch (Exception e) { e.printStackTrace(); } 
     } 

    } 

} 

....和相应的单元测试：

MessageTest.java

public class MessageTest extends TestCase { 

    Message msg; 

    static final String testContents = "hello there"; 

    public void setUp() { 

     // mock Socket 
     Mockit.redefineMethods(Socket.class, new Object() 
     { 

      ByteArrayOutputStream output = new ByteArrayOutputStream(); 
      ByteArrayInputStream input = new ByteArrayInputStream(); 

      public OutputStream getOutputStream() 
      { 
       return output; 
      } 

      public InputStream getInputStream() 
      { 
       return input; 
      } 


     }); 

     // NB 
     // some code removed here for simplicity 
     // which uses JMockit to overrides the Client object and give it a fake hostname and address 

     Client[] testClient = { new Client() }; 

     msg = new Message(testClient, testContents, this); 

    } 

    public void tearDown() { 
    } 

    public void testDispatch() { 

     // dispatch to client 
     msg.dispatch(); 


    } 
} 

Answer 1

注意，通过NIO API（java.nio）也可以在单个阻塞方法中实现多个消息（多播）的发送，而无需创建新线程。虽然NIO非常复杂。

我将首先编写测试，使用测试定义的StatusListener实现将所有更新事件存储在列表中。当dispatch（）方法返回时，测试可以对事件列表的状态执行断言。

使用线程或NIO是Message类的实现细节。因此，除非您不介意将测试与此实现细节相结合，否则我会建议引入一个辅助类，该类将负责发送多个异步消息并在任何异步回复时通知消息对象。然后，你可以在单元测试中模拟助手类，而不用将它们耦合到线程或NIO。

我成功实施了向一个客户端发送消息的情况的测试。我也做了一些改变原来的生产代码，如下所示：

public class Message 
{ 
    private static final int CLIENT_PORT = 8000; 

    // Externally provided: 
    private final Client[] to; 
    private final String contents; 
    private final StatusListener listener; 

    // Internal state: 
    private final List<Socket> clientConnections; 
    private final BlockingQueue<Status> statusQueue; 

    public Message(Client[] to, String contents, StatusListener listener) 
    { 
     this.to = to; 
     this.contents = contents; 
     this.listener = listener; 

     // Keep a list of socket references so that all threads can be closed: 
     clientConnections = Collections.synchronizedList(new ArrayList<Socket>()); 

     // Initialise the statusQ for threads to report message status: 
     statusQueue = new ArrayBlockingQueue<Status>(to.length * 3); 
    } 

    public void dispatch() 
    { 
     // Dispatch to each client individually and wait for confirmation: 
     sendContentsToEachClientAsynchronously(); 

     Status statusChangeReceived; 

     do { 
     try { 
      // Now, monitor queue and empty the queue as it fills up (consumer): 
      statusChangeReceived = statusQueue.take(); 
     } 
     catch (InterruptedException ignore) { 
      break; 
     } 
     } 
     while (listener.updateStatus(statusChangeReceived)); 

     closeRemainingClientConnections(); 
    } 

    private void closeRemainingClientConnections() 
    { 
     for (Socket connection : clientConnections) { 
     try { 
      connection.close(); 
     } 
     catch (IOException ignore) { 
      // OK 
     } 
     } 

     clientConnections.clear(); 
    } 

    private void sendContentsToEachClientAsynchronously() 
    { 
     for (Client client : to) { 
     System.out.println("Started new thread"); 
     new Thread(new MessageDispatcher(client)).start(); 
     } 
    } 

    // One MessageDispatcher per client. 
    private final class MessageDispatcher implements Runnable 
    { 
     private final Client client; 

     MessageDispatcher(Client client) { this.client = client; } 

     public void run() 
     { 
     try { 
      communicateWithClient(); 
     } 
     catch (IOException e) { 
      throw new RuntimeException(e); 
     } 
     } 

     private void communicateWithClient() throws IOException 
     { 
     // Open connection to client: 
     Socket connection = new Socket(client.getAddress(), CLIENT_PORT); 

     try { 
      // Add client connection to synchronized list: 
      clientConnections.add(connection); 

      sendMessage(connection.getOutputStream()); 
      readRequiredReceipts(connection.getInputStream()); 
     } 
     finally { 
      connection.close(); 
     } 
     } 

     // Send message and confirm dispatch. 
     private void sendMessage(OutputStream output) 
     { 
     PrintWriter out = new PrintWriter(output, true); 

     out.println(contents); 
     statusQueue.add(new Status(client, "DISPATCHED")); 
     } 

     private void readRequiredReceipts(InputStream input) throws IOException 
     { 
     BufferedReader in = new BufferedReader(new InputStreamReader(input)); 

     // Wait for display receipt: 
     in.readLine(); 
     statusQueue.add(new Status(client, "DISPLAYED")); 

     // Wait for read receipt: 
     in.readLine(); 
     statusQueue.add(new Status(client, "READ")); 
     } 
    } 
}

public final class MessageTest extends JMockitTest 
{ 
    static final String testContents = "hello there"; 
    static final String[] expectedEvents = {"DISPATCHED", "DISPLAYED", "READ"}; 

    @Test 
    public void testSendMessageToSingleClient() 
    { 
     final Client theClient = new Client("client1"); 
     Client[] testClient = {theClient}; 

     new MockUp<Socket>() 
     { 
     @Mock(invocations = 1) 
     void $init(String host, int port) 
     { 
      assertEquals(theClient.getAddress(), host); 
      assertTrue(port > 0); 
     } 

     @Mock(invocations = 1) 
     public OutputStream getOutputStream() { return new ByteArrayOutputStream(); } 

     @Mock(invocations = 1) 
     public InputStream getInputStream() 
     { 
      return new ByteArrayInputStream("reply1\nreply2\n".getBytes()); 
     } 

     @Mock(minInvocations = 1) void close() {} 
     }; 

     StatusListener listener = new MockUp<StatusListener>() 
     { 
     int eventIndex; 

     @Mock(invocations = 3) 
     boolean updateStatus(Status status) 
     { 
      assertSame(theClient, status.getClient()); 
      assertEquals(expectedEvents[eventIndex++], status.getEvent()); 
      return eventIndex < expectedEvents.length; 
     } 
     }.getMockInstance(); 

     new Message(testClient, testContents, listener).dispatch(); 
    } 
}

的JMockit测试上面采用了全新的MockUp类，在最新版本尚未公布。不过，它可以替换为Mockit.setUpMock(Socket.class, new Object() { ... })。

Answer 2

可能不是重新定义方法getOutputStream和getInputStream，而是可以在您的Message类中使用AbstractFactory来创建输出和输入流。在正常的操作中，工厂将使用一个插座来实现这一点。但是，为了进行测试，请为其提供一个工厂，让您选择它的流。这样你就可以更好地控制发生了什么。