java类io.netty.channel.DefaultChannelPromise的实例源码

@Test(expected = RuntimeException.class)
public void testIdentifyWhenWriteFailsAndChannelIsActiveClosesChannelAndThrows() throws Exception {
    DefaultChannelPromise promise = new DefaultChannelPromise(channel, eventExecutors.next());
    promise.setFailure(new IOException("test"));

    when(channel.writeAndFlush(any())).thenReturn(promise);
    when(channel.isActive()).thenReturn(true);
    when(channel.close()).thenReturn(promise);

    Admin admin = new Admin();
    admin.setBoxId("test");
    admin.setAdminCommand(AdminCommand.IDENTIFY);

    try {
        clientSession.identify(admin);
    } finally {
        verify(channel).writeAndFlush(admin);
        verify(channel).close();

        assertTrue(clientSession.isClosed());
    }
}

@Test(expected = RuntimeException.class)
public void testIdentifyWhenWriteFailsAndChannelIsInactiveSetsClosedState() throws Exception {
    DefaultChannelPromise promise = new DefaultChannelPromise(channel, eventExecutors.next());
    promise.setFailure(new IOException("test"));

    when(channel.writeAndFlush(any())).thenReturn(promise);
    when(channel.isActive()).thenReturn(false);

    Admin admin = new Admin();
    admin.setBoxId("test");
    admin.setAdminCommand(AdminCommand.IDENTIFY);

    try {
        clientSession.identify(admin);
    } finally {
        verify(channel).writeAndFlush(admin);
        verify(channel, times(0)).close();

        assertTrue(clientSession.isClosed());
    }
}

@Test
public void testDestroysClosesChannelAndDestroysTheWindow() throws Exception {
    DefaultChannelPromise promise = new DefaultChannelPromise(channel, eventExecutors.next());
    promise.setSuccess();

    when(channel.isActive()).thenReturn(false);
    when(channel.close()).thenReturn(promise);

    WindowFuture windowFuture = clientSession.getWindow().offer(UUID.randomUUID(), new Sms(), 5000);
    clientSession.destroy();

    verify(channel, times(0)).close();
    assertNull("Session handler must be null after destruction", clientSession.getSessionHandler());
    assertTrue("The outstanding requests should be canceled", windowFuture.isCancelled());

}

@Test(expected = DuplicateKeyException.class)
public void testSendSmsReturnsFailedFutureWhenOfferToWindowFails() throws Exception {
    DefaultChannelPromise promise = new DefaultChannelPromise(channel, eventExecutors.next());
    promise.setSuccess();

    when(channel.writeAndFlush(any())).thenReturn(promise);

    Sms sms = new Sms();
    sms.setId(UUID.randomUUID());
    sms.setBoxId("test box");

    //add the sms so the next offer fails
    clientSession.getWindow().offer(sms.getId(), sms, 5000);

    WindowFuture<Sms, Ack> future = clientSession.sendSms(sms, 5000);
    assertFalse(future.isCancelled());

    assertSame(sms, future.getRequest());

    Futures.getChecked(future, DuplicateKeyException.class);
}

@Test
public void testSendSmsAndWaitReturnsCorrectResponse() throws Exception {
    final DefaultChannelPromise promise = new DefaultChannelPromise(channel, eventExecutors.next());
    promise.setSuccess();

    when(channel.writeAndFlush(any())).thenReturn(promise);

    final Sms sms = new Sms();
    sms.setId(UUID.randomUUID());
    sms.setBoxId("test box");

    final Ack expectedResponse = new Ack();

    scheduledExecutorService.schedule(new Runnable() {
        @Override
        public void run() {
            clientSession.getWindow().complete(sms.getId(), expectedResponse);
        }
    }, 100, TimeUnit.MILLISECONDS);

    final Ack response = clientSession.sendSmsAndWait(sms, 5000);
    assertSame(expectedResponse, response);
}

@Test
public void testSendSmsAndWaitThrowsWhenOfferToWindowFails() throws Exception {
    DefaultChannelPromise promise = new DefaultChannelPromise(channel, eventExecutors.next());
    promise.setSuccess();

    when(channel.writeAndFlush(any())).thenReturn(promise);

    Sms sms = new Sms();
    sms.setId(UUID.randomUUID());
    sms.setBoxId("test box");

    //add the sms so the next offer fails
    clientSession.getWindow().offer(sms.getId(), sms, 5000);

    try {
        clientSession.sendSmsAndWait(sms, 5000);
    } catch (ExecutionException e) {
        assertTrue(e.getCause() instanceof DuplicateKeyException);
    }
}

@Test
public void testSendSmsAndWaitThrowsWhenWriteFails() throws Exception {
    DefaultChannelPromise promise = new DefaultChannelPromise(channel, eventExecutors.next());
    promise.setFailure(new IOException());

    when(channel.writeAndFlush(any())).thenReturn(promise);

    Sms sms = new Sms();
    sms.setId(UUID.randomUUID());
    sms.setBoxId("test box");

    try {
        clientSession.sendSmsAndWait(sms, 5000);
    } catch (ExecutionException e) {
        assertTrue(e.getCause() instanceof IOException);
    }
}

/**
 * Validates successful {@link WebSessionResources#close()} with valid CloseFuture and other parameters.
 * @throws Exception
 */
@Test
public void testChannelPromiseWithValidExecutor() throws Exception {
  try {
    EventExecutor mockExecutor = mock(EventExecutor.class);
    ChannelPromise closeFuture = new DefaultChannelPromise(null, mockExecutor);
    webSessionResources = new WebSessionResources(mock(BufferAllocator.class), mock(SocketAddress.class), mock
        (UserSession.class), closeFuture);
    webSessionResources.close();
    verify(webSessionResources.getAllocator()).close();
    verify(webSessionResources.getSession()).close();
    verify(mockExecutor).inEventLoop();
    verify(mockExecutor).execute(any(Runnable.class));
    assertTrue(webSessionResources.getCloseFuture() == null);
    assertTrue(!listenerComplete);
  } catch (Exception e) {
    fail();
  }
}

/**
 * Validates double call to {@link WebSessionResources#close()} doesn't throw any exception.
 * @throws Exception
 */
@Test
public void testDoubleClose() throws Exception {
  try {
    ChannelPromise closeFuture = new DefaultChannelPromise(null, mock(EventExecutor.class));
    webSessionResources = new WebSessionResources(mock(BufferAllocator.class), mock(SocketAddress.class), mock
        (UserSession.class), closeFuture);
    webSessionResources.close();

    verify(webSessionResources.getAllocator()).close();
    verify(webSessionResources.getSession()).close();
    assertTrue(webSessionResources.getCloseFuture() == null);

    webSessionResources.close();
  } catch (Exception e) {
    fail();
  }
}

@Test
public void itClosesTheUnderlyingChannel() {
  DefaultChannelPromise channelPromise = new DefaultChannelPromise(channel, ImmediateEventExecutor.INSTANCE);
  when(channel.close()).thenReturn(channelPromise);

  CompletableFuture<Void> f = session.close();
  channelPromise.setSuccess();

  assertThat(f.isDone());
}

private void assertExceptionsFiredOnFailure() throws Exception {
  // get the listener added when the channel was written to
  ArgumentCaptor<ChannelFutureListener> captor = ArgumentCaptor.forClass(ChannelFutureListener.class);
  verify(writeFuture, atLeast(1)).addListener(captor.capture());
  ChannelFutureListener addedListener = captor.getValue();

  // tell the listener the write failed
  DefaultChannelPromise promise = new DefaultChannelPromise(channel, ImmediateEventExecutor.INSTANCE);
  promise.setFailure(new Exception());
  addedListener.operationComplete(promise);

  verify(pipeline).fireExceptionCaught(promise.cause());
}

/** Reset the channel mock and set basic method call expectations */
void resetChannel() {
    reset(channel);
    expect(channel.newPromise()).andAnswer(new IAnswer<ChannelPromise>() {
        @Override
        public ChannelPromise answer() throws Throwable {
            return new DefaultChannelPromise(channel);
        }
    }).anyTimes();
    eventLoop = new TestEventLoop();
    expect(channel.eventLoop()).andReturn(eventLoop).anyTimes();
    expect(channel.pipeline()).andReturn(pipeline).anyTimes();
    expect(channel.remoteAddress()).andReturn(null).anyTimes();
}

/** Reset the channel mock and set basic method call expectations */
  void resetChannel() {
      reset(channel);
      expect(channel.newPromise()).andAnswer(new IAnswer<ChannelPromise>() {
    @Override
    public ChannelPromise answer() throws Throwable {
        return new DefaultChannelPromise(channel);
    }
}).anyTimes();
eventLoop = new TestEventLoop();
expect(channel.eventLoop()).andReturn(eventLoop).anyTimes();
      expect(channel.pipeline()).andReturn(pipeline).anyTimes();
      expect(channel.remoteAddress()).andReturn(InetSocketAddress.createUnresolved("1.1.1.1", 80)).anyTimes();
  }

/** Reset the channel mock and set basic method call expectations */
void resetChannel() {
    reset(channel);
    expect(channel.newPromise()).andAnswer(new IAnswer<ChannelPromise>() {
        @Override
        public ChannelPromise answer() throws Throwable {
            return new DefaultChannelPromise(channel);
        }
    }).anyTimes();
    eventLoop = new TestEventLoop();
    expect(channel.eventLoop()).andReturn(eventLoop).anyTimes();
    expect(channel.pipeline()).andReturn(pipeline).anyTimes();
    expect(channel.remoteAddress()).andReturn(null).anyTimes();
}

/** Reset the channel mock and set basic method call expectations */
  void resetChannel() {
      reset(channel);
      expect(channel.newPromise()).andAnswer(new IAnswer<ChannelPromise>() {
    @Override
    public ChannelPromise answer() throws Throwable {
        return new DefaultChannelPromise(channel);
    }
}).anyTimes();
eventLoop = new TestEventLoop();
expect(channel.eventLoop()).andReturn(eventLoop).anyTimes();
      expect(channel.pipeline()).andReturn(pipeline).anyTimes();
      expect(channel.remoteAddress()).andReturn(InetSocketAddress.createUnresolved("1.1.1.1", 80)).anyTimes();
  }

/** Reset the channel mock and set basic method call expectations */
void resetChannel() {
    reset(channel);
    expect(channel.newPromise()).andAnswer(new IAnswer<ChannelPromise>() {
        @Override
        public ChannelPromise answer() throws Throwable {
            return new DefaultChannelPromise(channel);
        }
    }).anyTimes();
    eventLoop = new TestEventLoop();
    expect(channel.eventLoop()).andReturn(eventLoop).anyTimes();
    expect(channel.pipeline()).andReturn(pipeline).anyTimes();
    expect(channel.remoteAddress()).andReturn(null).anyTimes();
}

/** Reset the channel mock and set basic method call expectations */
  void resetChannel() {
      reset(channel);
      expect(channel.newPromise()).andAnswer(new IAnswer<ChannelPromise>() {
    @Override
    public ChannelPromise answer() throws Throwable {
        return new DefaultChannelPromise(channel);
    }
}).anyTimes();
eventLoop = new TestEventLoop();
expect(channel.eventLoop()).andReturn(eventLoop).anyTimes();
      expect(channel.pipeline()).andReturn(pipeline).anyTimes();
      expect(channel.remoteAddress()).andReturn(InetSocketAddress.createUnresolved("1.1.1.1", 80)).anyTimes();
  }

/** Reset the channel mock and set basic method call expectations */
void resetChannel() {
    reset(channel);
    expect(channel.newPromise()).andAnswer(new IAnswer<ChannelPromise>() {
        @Override
        public ChannelPromise answer() throws Throwable {
            return new DefaultChannelPromise(channel);
        }
    }).anyTimes();
    eventLoop = new TestEventLoop();
    expect(channel.eventLoop()).andReturn(eventLoop).anyTimes();
    expect(channel.pipeline()).andReturn(pipeline).anyTimes();
    expect(channel.remoteAddress()).andReturn(null).anyTimes();
}

/** Reset the channel mock and set basic method call expectations */
  void resetChannel() {
      reset(channel);
      expect(channel.newPromise()).andAnswer(new IAnswer<ChannelPromise>() {
    @Override
    public ChannelPromise answer() throws Throwable {
        return new DefaultChannelPromise(channel);
    }
}).anyTimes();
eventLoop = new TestEventLoop();
expect(channel.eventLoop()).andReturn(eventLoop).anyTimes();
      expect(channel.pipeline()).andReturn(pipeline).anyTimes();
      expect(channel.remoteAddress()).andReturn(InetSocketAddress.createUnresolved("1.1.1.1", 80)).anyTimes();
  }

/** Reset the channel mock and set basic method call expectations */
void resetChannel() {
    reset(channel);
    expect(channel.newPromise()).andAnswer(new IAnswer<ChannelPromise>() {
        @Override
        public ChannelPromise answer() throws Throwable {
            return new DefaultChannelPromise(channel);
        }
    }).anyTimes();
    eventLoop = new TestEventLoop();
    expect(channel.eventLoop()).andReturn(eventLoop).anyTimes();
    expect(channel.pipeline()).andReturn(pipeline).anyTimes();
    expect(channel.remoteAddress()).andReturn(null).anyTimes();
}

/** Reset the channel mock and set basic method call expectations */
  void resetChannel() {
      reset(channel);
      expect(channel.newPromise()).andAnswer(new IAnswer<ChannelPromise>() {
    @Override
    public ChannelPromise answer() throws Throwable {
        return new DefaultChannelPromise(channel);
    }
}).anyTimes();
eventLoop = new TestEventLoop();
expect(channel.eventLoop()).andReturn(eventLoop).anyTimes();
      expect(channel.pipeline()).andReturn(pipeline).anyTimes();
      expect(channel.remoteAddress()).andReturn(InetSocketAddress.createUnresolved("1.1.1.1", 80)).anyTimes();
  }

/** Reset the channel mock and set basic method call expectations */
void resetChannel() {
    reset(channel);
    expect(channel.newPromise()).andAnswer(new IAnswer<ChannelPromise>() {
        @Override
        public ChannelPromise answer() throws Throwable {
            return new DefaultChannelPromise(channel);
        }
    }).anyTimes();
    eventLoop = new TestEventLoop();
    expect(channel.eventLoop()).andReturn(eventLoop).anyTimes();
    expect(channel.pipeline()).andReturn(pipeline).anyTimes();
    expect(channel.remoteAddress()).andReturn(null).anyTimes();
}

/** Reset the channel mock and set basic method call expectations */
  void resetChannel() {
      reset(channel);
      expect(channel.newPromise()).andAnswer(new IAnswer<ChannelPromise>() {
    @Override
    public ChannelPromise answer() throws Throwable {
        return new DefaultChannelPromise(channel);
    }
}).anyTimes();
eventLoop = new TestEventLoop();
expect(channel.eventLoop()).andReturn(eventLoop).anyTimes();
      expect(channel.pipeline()).andReturn(pipeline).anyTimes();
      expect(channel.remoteAddress()).andReturn(InetSocketAddress.createUnresolved("1.1.1.1", 80)).anyTimes();
  }

@Test
public void testSendAck() throws Exception {
    DefaultChannelPromise promise = new DefaultChannelPromise(channel, eventExecutors.next());
    promise.setSuccess();

    when(channel.writeAndFlush(any())).thenReturn(promise);

    Ack ack = mock(Ack.class);
    assertSame(promise, clientSession.sendAck(ack));

    verify(channel).writeAndFlush(ack);
}

@Test(expected = IllegalStateException.class)
public void testIdentifyWhenCommandIsNotIdentifyThrows() throws Exception {
    DefaultChannelPromise promise = new DefaultChannelPromise(channel, eventExecutors.next());
    promise.setSuccess();

    when(channel.writeAndFlush(any())).thenReturn(promise);

    Admin admin = new Admin();
    admin.setBoxId("test");
    admin.setAdminCommand(AdminCommand.RESTART);

    clientSession.identify(admin);
}

@Test
public void testCloseWhenChannelIsActiveClosesChannel() throws Exception {
    DefaultChannelPromise promise = new DefaultChannelPromise(channel, eventExecutors.next());
    promise.setSuccess();

    when(channel.isActive()).thenReturn(true);
    when(channel.close()).thenReturn(promise);

    clientSession.close();

    verify(channel).close();
}

@Test
public void testCloseWhenChannelIsInactiveClosesChannel() throws Exception {
    DefaultChannelPromise promise = new DefaultChannelPromise(channel, eventExecutors.next());
    promise.setSuccess();

    when(channel.isActive()).thenReturn(false);
    when(channel.close()).thenReturn(promise);

    clientSession.close();

    verify(channel, times(0)).close();
}

@Test
public void testSendSmsSetsUUIDAndBoxIdWhenNull() throws Exception {
    DefaultChannelPromise promise = new DefaultChannelPromise(channel, eventExecutors.next());
    promise.setSuccess();

    when(channel.writeAndFlush(any())).thenReturn(promise);

    Sms sms = new Sms("from", "to", "date", SmsType.MOBILE_TERMINATED_PUSH, DataCoding.DC_8BIT);

    clientSession.sendSms(sms, 5000);

    assertNotNull(sms.getId());
    assertSame(clientSessionConfiguration.getClientId(), sms.getBoxId());
    verify(channel).writeAndFlush(sms);
}

@Test(expected = IOException.class)
public void testSendSmsReturnsFailedFutureWhenWriteFails() throws Exception {
    DefaultChannelPromise promise = new DefaultChannelPromise(channel, eventExecutors.next());
    promise.setFailure(new IOException());

    when(channel.writeAndFlush(any())).thenReturn(promise);

    Sms sms = new Sms();
    sms.setId(UUID.randomUUID());
    sms.setBoxId("test box");

    WindowFuture<Sms, Ack> future = clientSession.sendSms(sms, 5000);

    Futures.getChecked(future, IOException.class);
}

@Test(expected = CancellationException.class)
public void testSendSmsReturnsFailedFutureWhenWriteIsCancelled() throws Exception {
    DefaultChannelPromise promise = new DefaultChannelPromise(channel, eventExecutors.next());
    promise.cancel(true);

    when(channel.writeAndFlush(any())).thenReturn(promise);

    Sms sms = new Sms();
    sms.setId(UUID.randomUUID());
    sms.setBoxId("test box");

    WindowFuture<Sms, Ack> future = clientSession.sendSms(sms, 5000);

    Futures.getUnchecked(future);
}