AMQPSessionCallback implements SessionCallback { public void flow(final SimpleString address, Runnable runnable) { try { PagingManager pagingManager = manager.getServer().getPagingManager(); if (address == null) { pagingManager.checkMemory(runnable); } else { final PagingStore store = manager.getServer().getPagingManager().getPageStore(address); if (store != null) { store.checkMemory(runnable); } else { runnable.run(); } } } catch (Exception e) { throw new RuntimeException(e); } } AMQPSessionCallback(AMQPConnectionCallback protonSPI,
                              ProtonProtocolManager manager,
                              AMQPConnectionContext connection,
                              Connection transportConnection,
                              Executor executor,
                              OperationContext operationContext); StorageManager getStorageManager(); CoreMessageObjectPools getCoreMessageObjectPools(); @Override boolean isWritable(ReadyListener callback, Object protocolContext); void withinSessionExecutor(Runnable run); void addCloseable(Closeable closeable); void withinContext(Runnable run); void afterIO(IOCallback ioCallback); @Override void browserFinished(ServerConsumer consumer); @Override boolean supportsDirectDelivery(); void init(AMQPSessionContext protonSession, SASLResult saslResult); @Override void afterDelivery(); void start(); Object createSender(ProtonServerSenderContext protonSender,
                              SimpleString queue,
                              String filter,
                              boolean browserOnly); void startSender(Object brokerConsumer); void createTemporaryQueue(SimpleString queueName, RoutingType routingType); void createTemporaryQueue(SimpleString address,
                                    SimpleString queueName,
                                    RoutingType routingType,
                                    SimpleString filter); void createUnsharedDurableQueue(SimpleString address,
                                          RoutingType routingType,
                                          SimpleString queueName,
                                          SimpleString filter); void createSharedDurableQueue(SimpleString address,
                                        RoutingType routingType,
                                        SimpleString queueName,
                                        SimpleString filter); void createSharedVolatileQueue(SimpleString address,
                                         RoutingType routingType,
                                         SimpleString queueName,
                                         SimpleString filter); QueueQueryResult queueQuery(SimpleString queueName, RoutingType routingType, boolean autoCreate); boolean checkAddressAndAutocreateIfPossible(SimpleString address, RoutingType routingType); AddressQueryResult addressQuery(SimpleString addressName,
                                          RoutingType routingType,
                                          boolean autoCreate); void closeSender(final Object brokerConsumer); String tempQueueName(); void close(); void ack(Transaction transaction, Object brokerConsumer, Message message); void cancel(Object brokerConsumer, Message message, boolean updateCounts); void reject(Object brokerConsumer, Message message); void resumeDelivery(Object consumer); AMQPStandardMessage createStandardMessage(Delivery delivery, ReadableBuffer data); void serverSend(ProtonServerReceiverContext context,
                          Transaction transaction,
                          Receiver receiver,
                          Delivery delivery,
                          SimpleString address,
                          RoutingContext routingContext,
                          AMQPMessage message); void flow(final SimpleString address,
                    Runnable runnable); void deleteQueue(SimpleString queueName); void resetContext(OperationContext oldContext); OperationContext recoverContext(); @Override void sendProducerCreditsMessage(int credits, SimpleString address); @Override boolean updateDeliveryCountAfterCancel(ServerConsumer consumer, MessageReference ref, boolean failed); @Override void sendProducerCreditsFailMessage(int credits, SimpleString address); @Override int sendMessage(MessageReference ref, Message message, ServerConsumer consumer, int deliveryCount); @Override int sendLargeMessage(MessageReference ref,
                               Message message,
                               ServerConsumer consumer,
                               long bodySize,
                               int deliveryCount); @Override int sendLargeMessageContinuation(ServerConsumer consumer,
                                           byte[] body,
                                           boolean continues,
                                           boolean requiresResponse); @Override void closed(); @Override void disconnect(ServerConsumer consumer, SimpleString queueName); @Override boolean hasCredits(ServerConsumer consumer); @Override Transaction getCurrentTransaction(); Transaction getTransaction(Binary txid, boolean remove); Binary newTransaction(); SimpleString getMatchingQueue(SimpleString address, RoutingType routingType); SimpleString getMatchingQueue(SimpleString address,
                                        SimpleString queueName,
                                        RoutingType routingType); AddressInfo getAddress(SimpleString address); void removeTemporaryQueue(SimpleString address); RoutingType getDefaultRoutingType(SimpleString address); void check(SimpleString address, CheckType checkType, SecurityAuth session); String invokeIncoming(AMQPMessage message, ActiveMQProtonRemotingConnection connection); String invokeOutgoing(AMQPMessage message, ActiveMQProtonRemotingConnection connection); void addProducer(ServerProducer serverProducer); void removeProducer(String name); void setTransactionHandler(ProtonTransactionHandler transactionHandler); Connection getTransportConnection(); ProtonTransactionHandler getTransactionHandler();  }
@Test public void testOfferProducerWithNoAddressDoesNotTopOffCreditAboveThreshold() { Mockito.when(manager.getServer()).thenReturn(server); Mockito.when(server.getPagingManager()).thenReturn(pagingManager); ArgumentCaptor<Runnable> argument = ArgumentCaptor.forClass(Runnable.class); AMQPSessionCallback session = new AMQPSessionCallback(protonSPI, manager, connection, transportConnection, executor, operationContext); Mockito.when(receiver.getCredit()).thenReturn(AMQP_LOW_CREDITS_DEFAULT + 1); session.flow(null, ProtonServerReceiverContext.createCreditRunnable(AMQP_CREDITS_DEFAULT, AMQP_LOW_CREDITS_DEFAULT, receiver, connection)); Mockito.verify(pagingManager).checkMemory(argument.capture()); assertNotNull(argument.getValue()); argument.getValue().run(); Mockito.verify(receiver, never()).getRemoteCredit(); Mockito.verify(receiver, never()).flow(anyInt()); }
@Test public void testOfferProducerWithNoAddressTopsOffCreditAtThreshold() { Mockito.when(manager.getServer()).thenReturn(server); Mockito.when(server.getPagingManager()).thenReturn(pagingManager); ArgumentCaptor<Runnable> argument = ArgumentCaptor.forClass(Runnable.class); AMQPSessionCallback session = new AMQPSessionCallback(protonSPI, manager, connection, transportConnection, executor, operationContext); Mockito.when(receiver.getCredit()).thenReturn(AMQP_LOW_CREDITS_DEFAULT); session.flow(null, ProtonServerReceiverContext.createCreditRunnable(AMQP_CREDITS_DEFAULT, AMQP_LOW_CREDITS_DEFAULT, receiver, connection)); Mockito.verify(pagingManager).checkMemory(argument.capture()); assertNotNull(argument.getValue()); argument.getValue().run(); Mockito.verify(receiver, never()).getRemoteCredit(); Mockito.verify(receiver).flow(AMQP_CREDITS_DEFAULT - AMQP_LOW_CREDITS_DEFAULT); }
@Test public void testOfferProducerWithAddressDoesNotTopOffCreditAboveThreshold() throws Exception { Mockito.when(manager.getServer()).thenReturn(server); Mockito.when(server.getPagingManager()).thenReturn(pagingManager); Mockito.when(pagingManager.getPageStore(any(SimpleString.class))).thenReturn(pagingStore); ArgumentCaptor<Runnable> argument = ArgumentCaptor.forClass(Runnable.class); AMQPSessionCallback session = new AMQPSessionCallback(protonSPI, manager, connection, transportConnection, executor, operationContext); Mockito.when(receiver.getCredit()).thenReturn(AMQP_LOW_CREDITS_DEFAULT + 1); session.flow(new SimpleString("test"), ProtonServerReceiverContext.createCreditRunnable(AMQP_CREDITS_DEFAULT, AMQP_LOW_CREDITS_DEFAULT, receiver, connection)); Mockito.verify(pagingStore).checkMemory(argument.capture()); assertNotNull(argument.getValue()); argument.getValue().run(); Mockito.verify(receiver, never()).getRemoteCredit(); Mockito.verify(receiver, never()).flow(anyInt()); }
@Test public void testOfferProducerWithAddressTopsOffCreditAtThreshold() throws Exception { Mockito.when(manager.getServer()).thenReturn(server); Mockito.when(server.getPagingManager()).thenReturn(pagingManager); Mockito.when(pagingManager.getPageStore(any(SimpleString.class))).thenReturn(pagingStore); ArgumentCaptor<Runnable> argument = ArgumentCaptor.forClass(Runnable.class); AMQPSessionCallback session = new AMQPSessionCallback(protonSPI, manager, connection, transportConnection, executor, operationContext); Mockito.when(receiver.getCredit()).thenReturn(AMQP_LOW_CREDITS_DEFAULT); session.flow(new SimpleString("test"), ProtonServerReceiverContext.createCreditRunnable(AMQP_CREDITS_DEFAULT, AMQP_LOW_CREDITS_DEFAULT, receiver, connection)); Mockito.verify(pagingStore).checkMemory(argument.capture()); assertNotNull(argument.getValue()); argument.getValue().run(); Mockito.verify(receiver, never()).getRemoteCredit(); Mockito.verify(receiver).flow(AMQP_CREDITS_DEFAULT - AMQP_LOW_CREDITS_DEFAULT); }
@Test public void testOfferProducerWithNoAddressDoesNotGrantNegativeCredit() { Mockito.when(manager.getServer()).thenReturn(server); Mockito.when(server.getPagingManager()).thenReturn(pagingManager); ArgumentCaptor<Runnable> argument = ArgumentCaptor.forClass(Runnable.class); AMQPSessionCallback session = new AMQPSessionCallback(protonSPI, manager, connection, transportConnection, executor, operationContext); Mockito.when(receiver.getCredit()).thenReturn(AMQP_LOW_CREDITS_DEFAULT); session.flow(null, ProtonServerReceiverContext.createCreditRunnable(1, AMQP_LOW_CREDITS_DEFAULT, receiver, connection)); Mockito.verify(pagingManager).checkMemory(argument.capture()); assertNotNull(argument.getValue()); argument.getValue().run(); Mockito.verify(receiver, never()).getRemoteCredit(); Mockito.verify(receiver, never()).flow(anyInt()); }
@Test public void testOfferProducerWithAddressDoesNotGrantNegativeCredit() throws Exception { Mockito.when(manager.getServer()).thenReturn(server); Mockito.when(server.getPagingManager()).thenReturn(pagingManager); Mockito.when(pagingManager.getPageStore(any(SimpleString.class))).thenReturn(pagingStore); ArgumentCaptor<Runnable> argument = ArgumentCaptor.forClass(Runnable.class); AMQPSessionCallback session = new AMQPSessionCallback(protonSPI, manager, connection, transportConnection, executor, operationContext); Mockito.when(receiver.getCredit()).thenReturn(AMQP_LOW_CREDITS_DEFAULT); session.flow(new SimpleString("test"), ProtonServerReceiverContext.createCreditRunnable(1, AMQP_LOW_CREDITS_DEFAULT, receiver, connection)); Mockito.verify(pagingStore).checkMemory(argument.capture()); assertNotNull(argument.getValue()); argument.getValue().run(); Mockito.verify(receiver, never()).getRemoteCredit(); Mockito.verify(receiver, never()).flow(anyInt()); }