JMSConnectionHandlingActor extends AbstractActor { private void handleRecoverSession(final AmqpClientActor.JmsRecoverSession recoverSession) { log.debug("Processing JmsRecoverSession message."); final ActorRef sender = getSender(); final ActorRef origin = recoverSession.getOrigin().orElse(null); final ActorRef self = getSelf(); recoverSession.getSession().ifPresent(session -> { try { session.close(); } catch (final JMSException e) { log.debug("Failed to close previous session, ignore."); } }); final Optional<javax.jms.Connection> connectionOptional = recoverSession.getConnection(); if (connectionOptional.isPresent()) { final JmsConnection jmsConnection = (JmsConnection) connectionOptional.get(); try { log.debug("Creating new JMS session."); final Session session = createSession(jmsConnection); log.debug("Creating consumers for new session."); final List<ConsumerData> consumers = createConsumers(session); final AmqpClientActor.JmsSessionRecovered r = new AmqpClientActor.JmsSessionRecovered(origin, session, consumers); sender.tell(r, self); log.debug("Session of connection <{}> recovered successfully.", connection.getId()); } catch (final ConnectionFailedException e) { sender.tell(new ImmutableConnectionFailure(origin, e, e.getMessage()), self); log.warning(e.getMessage()); } catch (final Exception e) { sender.tell(new ImmutableConnectionFailure(origin, e, e.getMessage()), self); log.error("Unexpected error: {}", e.getMessage()); } } else { log.info("Recovering session failed, no connection available."); sender.tell(new ImmutableConnectionFailure(origin, null, "Session recovery failed, no connection available."), self); } } @SuppressWarnings("unused") private  JMSConnectionHandlingActor(final Connection connection, final ExceptionListener exceptionListener,
            final JmsConnectionFactory jmsConnectionFactory); @Override Receive createReceive();  }
@Test public void handleRecoverSession() throws JMSException { new TestKit(actorSystem) {{ final Props props = JMSConnectionHandlingActor.props(connection, e -> {}, jmsConnectionFactory); final ActorRef connectionHandlingActor = watch(actorSystem.actorOf(props)); final TestProbe origin = TestProbe.apply(actorSystem); final JmsSession existingSession = Mockito.mock(JmsSession.class); connectionHandlingActor.tell( new AmqpClientActor.JmsRecoverSession(origin.ref(), mockConnection, existingSession), getRef()); final AmqpClientActor.JmsSessionRecovered recovered = expectMsgClass(AmqpClientActor.JmsSessionRecovered.class); assertThat(recovered.getOrigin()).contains(origin.ref()); assertThat(recovered.getSession()).isSameAs(mockSession); verify(existingSession).close(); verify(mockConnection).createSession(anyInt()); verify(mockSession, times(connection.getSources() .stream() .mapToInt(s -> s.getAddresses().size() * s.getConsumerCount()) .sum())).createConsumer(any()); }}; }