/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.activemq.artemis.tests.integration.amqp;

import javax.jms.JMSConsumer;
import javax.jms.JMSContext;
import javax.jms.JMSProducer;
import javax.jms.Message;
import javax.jms.Queue;
import javax.jms.Session;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.LongAdder;

import org.apache.qpid.jms.JmsConnectionFactory;
import org.junit.Assert;
import org.junit.Test;

public class DrainTimeoutTest extends AmqpClientTestSupport {

   final int NUMBER_OF_MESSAGES = 1000;

   @Test(timeout = 300_000)
   public void testFlowControl() throws Exception {
      final AtomicInteger errors = new AtomicInteger(0);
      final String queueName = getQueueName();
      JmsConnectionFactory connectionFactory =
         new JmsConnectionFactory(
            "amqp://localhost:5672?jms.prefetchPolicy.all=1&jms.connectTimeout=60000&amqp.drainTimeout=1000");
      LongAdder sendCount = new LongAdder();
      LongAdder consumeCount = new LongAdder();
      Thread consumerThread =
         new Thread(
            () -> {
               try (JMSContext listenerContext =
                       connectionFactory.createContext(Session.AUTO_ACKNOWLEDGE)) {
                  try (JMSConsumer consumer =
                          listenerContext.createConsumer(
                             listenerContext.createQueue(queueName))) {
                     while (!Thread.interrupted()) {
                        while (true) {
                           if (consumer.receiveNoWait() == null) {
                              break;
                           }
                           consumeCount.increment();
                           if (consumeCount.sum() == NUMBER_OF_MESSAGES) {
                              return;
                           }
                        }
                     }
                  }
               } catch (Exception e) {
                  e.printStackTrace(System.out);
                  errors.incrementAndGet();
               }
            });
      consumerThread.start();
      try (JMSContext context = connectionFactory.createContext(Session.AUTO_ACKNOWLEDGE)) {
         final Message message = context.createMessage();
         message.setStringProperty("selector", "dude");
         JMSProducer producer = context.createProducer();
         Queue queue = context.createQueue(queueName);
         while (sendCount.sum() < NUMBER_OF_MESSAGES && !Thread.interrupted()) {
            producer.send(queue, message);
            sendCount.increment();
            long sent = sendCount.sum();
         }
      }

      consumerThread.join();

      Assert.assertEquals(0, errors.get());
   }
}
