content: /*
   * 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.camel.quarkus.component.jt400.it;
  
  import java.nio.charset.StandardCharsets;
  import java.util.Collections;
  import java.util.Enumeration;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.LinkedList;
  import java.util.List;
  import java.util.Map;
  import java.util.Optional;
  import java.util.Set;
  import java.util.concurrent.TimeUnit;
  import java.util.stream.Collectors;
  
  import com.ibm.as400.access.AS400;
  import com.ibm.as400.access.DataQueue;
  import com.ibm.as400.access.DataQueueEntry;
  import com.ibm.as400.access.IFSFileInputStream;
  import com.ibm.as400.access.IFSKey;
  import com.ibm.as400.access.KeyedDataQueue;
  import com.ibm.as400.access.MessageQueue;
  import com.ibm.as400.access.QueuedMessage;
  import io.quarkus.test.common.QuarkusTestResourceLifecycleManager;
  import org.awaitility.Awaitility;
  import org.eclipse.microprofile.config.ConfigProvider;
  import org.hamcrest.Matchers;
  import org.jboss.logging.Logger;
  
  public class Jt400TestResource implements QuarkusTestResourceLifecycleManager {
      private static final Logger LOGGER = Logger.getLogger(Jt400TestResource.class);
  
      public static enum RESOURCE_TYPE {
          messageQueue,
          keyedDataQue,
          lifoQueueu,
          replyToQueueu;
      }
  
      private static final String JT400_URL = ConfigProvider.getConfig().getValue("cq.jt400.url", String.class);
      private static final String JT400_USERNAME = ConfigProvider.getConfig().getValue("cq.jt400.username", String.class);
      private static final String JT400_PASSWORD = ConfigProvider.getConfig().getValue("cq.jt400.password", String.class);
      private static final String JT400_LIBRARY = ConfigProvider.getConfig().getValue("cq.jt400.library", String.class);
      private static final String JT400_MESSAGE_QUEUE = ConfigProvider.getConfig().getValue("cq.jt400.message-queue",
              String.class);
      private static final String JT400_REPLY_TO_MESSAGE_QUEUE = ConfigProvider.getConfig().getValue(
              "cq.jt400.message-replyto-queue",
              String.class);
      private static final String JT400_LIFO_QUEUE = ConfigProvider.getConfig().getValue("cq.jt400.lifo-queue",
              String.class);
      private static final String JT400_KEYED_QUEUE = ConfigProvider.getConfig().getValue("cq.jt400.keyed-queue", String.class);
      private static final Optional<String> JT400_LOCK_FILE = ConfigProvider.getConfig().getOptionalValue("cq.jt400.lock-file",
              String.class);
  
      //depth of repetitive reads for lifo queue clearing
      private final static int CLEAR_DEPTH = 100;
      //10 minute timeout to obtain a log for the tests execution
      private final static int LOCK_TIMEOUT = 600000;
  
      private static AS400 lockAs400;
  
      @Override
      public Map<String, String> start() {
          //no need to start, as the instance already exists
          return Collections.emptyMap();
      }
  
      @Override
      public void stop() {
          //no need to unlock, once the as400 connection is release, the lock is released
          if (lockAs400 != null) {
              lockAs400.close();
          }
      }
  
      private static String getObjectPath(String object) {
          return String.format("/QSYS.LIB/%s.LIB/%s", JT400_LIBRARY, object);
      }
  
      public static Jt400ClientHelper CLIENT_HELPER = new Jt400ClientHelper() {
  
          private boolean cleared = false;
          Map<RESOURCE_TYPE, Set<Object>> toRemove = new HashMap<>();
  
          IFSFileInputStream lockFile;
          IFSKey lockKey;
  
          @Override
          public QueuedMessage peekReplyToQueueMessage(String msg) throws Exception {
              return getQueueMessage(JT400_REPLY_TO_MESSAGE_QUEUE, msg);
          }
  
          private QueuedMessage getQueueMessage(String queue, String msg) throws Exception {
              AS400 as400 = createAs400();
              try {
                  MessageQueue messageQueue = new MessageQueue(as400,
                          getObjectPath(queue));
                  Enumeration<QueuedMessage> msgs = messageQueue.getMessages();
  
                  while (msgs.hasMoreElements()) {
                      QueuedMessage queuedMessage = msgs.nextElement();
  
                      if (msg.equals(queuedMessage.getText())) {
                          return queuedMessage;
                      }
                  }
                  return null;
  
              } finally {
                  as400.close();
              }
          }
  
          @Override
          public void registerForRemoval(RESOURCE_TYPE type, Object value) {
              if (toRemove.containsKey(type)) {
                  toRemove.get(type).add(value);
              } else {
                  Set<Object> set = new HashSet<>();
                  set.add(value);
                  toRemove.put(type, set);
              }
          }
  
          @Override
          public boolean clear() throws Exception {
  
              //clear only once
              if (cleared) {
                  return false;
              }
  
              try (AS400 as400 = createAs400()) {
                  //reply-to queue
                  new MessageQueue(as400, getObjectPath(JT400_REPLY_TO_MESSAGE_QUEUE)).remove();
  
                  //message queue
                  new MessageQueue(as400, getObjectPath(JT400_MESSAGE_QUEUE)).remove();
  
                  //lifo queue
                  DataQueue dq = new DataQueue(as400, getObjectPath(JT400_LIFO_QUEUE));
                  for (int i = 01; i < CLEAR_DEPTH; i++) {
                      if (dq.read() == null) {
                          break;
                      }
                  }
  
                  //keyed queue
                  new KeyedDataQueue(as400, getObjectPath(JT400_KEYED_QUEUE)).clear();
              }
  
              return true;
          }
  
          /**
           * Locking is implemented via file locking, which is present in JTOpen.
           */
          @Override
          public void lock() throws Exception {
  
              //if no lock file is proposed, throw an error
              if (JT400_LOCK_FILE.isEmpty()) {
                  throw new IllegalStateException("No file for locking is provided.");
              }
  
              if (lockKey == null) {
                  lockAs400 = createAs400();
                  lockFile = new IFSFileInputStream(lockAs400, JT400_LOCK_FILE.get());
  
                  LOGGER.debug("Asked for lock.");
  
                  Awaitility.await().pollInterval(1, TimeUnit.SECONDS).atMost(LOCK_TIMEOUT, TimeUnit.SECONDS)
                          .until(() -> {
                              try {
                                  lockKey = lockFile.lock(1l);
                              } catch (Exception e) {
                                  //lock was not acquired
                                  return false;
                              }
                              LOGGER.debug("Acquired lock (for file `" + JT400_LOCK_FILE.get() + "`.");
                              return true;
  
                          },
                                  Matchers.is(true));
              }
          }
  
          @Override
          public String dumpQueues() throws Exception {
              AS400 as400 = createAs400();
              try {
                  StringBuilder sb = new StringBuilder();
  
                  sb.append("\n* MESSAGE QUEUE\n");
                  sb.append("\t" + Collections.list(new MessageQueue(as400, getObjectPath(JT400_MESSAGE_QUEUE)).getMessages())
                          .stream().map(mq -> mq.getText()).sorted().collect(Collectors.joining(", ")));
  
                  sb.append("\n* INQUIRY QUEUE\n");
                  sb.append("\t" + Collections
                          .list(new MessageQueue(as400, getObjectPath(JT400_REPLY_TO_MESSAGE_QUEUE)).getMessages())
                          .stream().map(mq -> mq.getText()).sorted().collect(Collectors.joining(", ")));
  
                  sb.append("\n* LIFO QUEUE\n");
                  DataQueue dq = new DataQueue(as400, getObjectPath(JT400_LIFO_QUEUE));
                  DataQueueEntry dqe;
                  List<byte[]> lifoMessages = new LinkedList<>();
                  List<String> lifoTexts = new LinkedList<>();
                  do {
                      dqe = dq.read();
                      if (dqe != null) {
                          lifoTexts.add(dqe.getString() + " (" + new String(dqe.getData(), StandardCharsets.UTF_8) + ")");
                          lifoMessages.add(dqe.getData());
                      }
                  } while (dqe != null);
  
                  //write back other messages in reverse order (it is a lifo)
                  Collections.reverse(lifoMessages);
                  for (byte[] msg : lifoMessages) {
                      dq.write(msg);
                  }
                  sb.append(lifoTexts.stream().collect(Collectors.joining(", ")));
  
                  //there is no api to list keyed queue, without knowledge of keys
                  return sb.toString();
  
              } finally {
                  as400.close();
              }
          }
  
          public void sendInquiry(String msg) throws Exception {
              AS400 as400 = createAs400();
              try {
                  new MessageQueue(as400, getObjectPath(JT400_REPLY_TO_MESSAGE_QUEUE)).sendInquiry(msg,
                          getObjectPath(JT400_REPLY_TO_MESSAGE_QUEUE));
              } finally {
                  as400.close();
              }
          }
      };
  
      private static AS400 createAs400() {
          return new AS400(JT400_URL, JT400_USERNAME, JT400_PASSWORD.toCharArray());
      }
  
  }
  
  interface Jt400ClientHelper {
  
      void registerForRemoval(Jt400TestResource.RESOURCE_TYPE type, Object value);
  
      QueuedMessage peekReplyToQueueMessage(String msg) throws Exception;
  
      void sendInquiry(String msg) throws Exception;
  
      //------------------- clear listeners ------------------------------
  
      boolean clear() throws Exception;
  
      //----------------------- locking
  
      void lock() throws Exception;
  
      String dumpQueues() throws Exception;
  
  }
  