/*
    * This file is part of Waarp Project (named also Waarp or GG).
    *
    *  Copyright (c) 2019, Waarp SAS, and individual contributors by the @author
    *  tags. See the COPYRIGHT.txt in the distribution for a full listing of
    * individual contributors.
    *
    *  All Waarp Project is free software: you can redistribute it and/or
    * modify it under the terms of the GNU General Public License as published by
   * the Free Software Foundation, either version 3 of the License, or (at your
   * option) any later version.
   *
   * Waarp is distributed in the hope that it will be useful, but WITHOUT ANY
   * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
   * A PARTICULAR PURPOSE. See the GNU General Public License for more details.
   *
   *  You should have received a copy of the GNU General Public License along with
   * Waarp . If not, see <http://www.gnu.org/licenses/>.
   */
  package org.waarp.openr66.client;
  
  import io.netty.channel.ChannelFuture;
  import org.waarp.common.database.exception.WaarpDatabaseException;
  import org.waarp.common.file.DataBlock;
  import org.waarp.common.logging.WaarpLoggerFactory;
  import org.waarp.openr66.commander.ClientRunner;
  import org.waarp.openr66.context.ErrorCode;
  import org.waarp.openr66.context.R66FiniteDualStates;
  import org.waarp.openr66.context.R66Result;
  import org.waarp.openr66.context.R66Session;
  import org.waarp.openr66.context.task.exception.OpenR66RunnerErrorException;
  import org.waarp.openr66.database.data.DbRule;
  import org.waarp.openr66.database.data.DbTaskRunner;
  import org.waarp.openr66.protocol.configuration.Configuration;
  import org.waarp.openr66.protocol.configuration.PartnerConfiguration;
  import org.waarp.openr66.protocol.exception.OpenR66DatabaseGlobalException;
  import org.waarp.openr66.protocol.exception.OpenR66Exception;
  import org.waarp.openr66.protocol.exception.OpenR66ProtocolNoConnectionException;
  import org.waarp.openr66.protocol.exception.OpenR66ProtocolNotYetConnectionException;
  import org.waarp.openr66.protocol.exception.OpenR66ProtocolPacketException;
  import org.waarp.openr66.protocol.exception.OpenR66ProtocolSystemException;
  import org.waarp.openr66.protocol.localhandler.LocalChannelReference;
  import org.waarp.openr66.protocol.localhandler.RetrieveRunner;
  import org.waarp.openr66.protocol.localhandler.packet.EndRequestPacket;
  import org.waarp.openr66.protocol.localhandler.packet.ErrorPacket;
  import org.waarp.openr66.protocol.localhandler.packet.RequestPacket;
  import org.waarp.openr66.protocol.networkhandler.NetworkTransaction;
  import org.waarp.openr66.protocol.utils.ChannelUtils;
  import org.waarp.openr66.protocol.utils.R66Future;
  
  /**
   * Class for Send Through client
   * <p>
   * This class does not included the real file transfer since it is up to the
   * business project to implement how
   * to read new data to be sent to the remote host. If an error occurs, no
   * transfer log is kept.
   * <p>
   * 1) Configuration must have been loaded<br>
   * <br>
   * 2) Pipeline and NetworkTransaction must have been initiated:<br>
   * <tt>     Configuration.configuration.pipelineInit();</tt><br>
   * <tt>     NetworkTransaction networkTransaction = new
   * NetworkTransaction();</tt><br>
   * <br>
   * 3) Prepare the request of transfer:<br>
   * <tt>     R66Future futureReq = new R66Future(true);</tt><br>
   * <tt>     SendThroughClient transaction = new SendThroughClient(futureReq,...);</tt><br>
   * <tt>     if (! transaction.initiateRequest()) { error }</tt><br>
   * <br>
   * 4) Once initiateRequest() gives true, you are ready to send the data in
   * through mode like:<br>
   * <tt>     byte[] data = readOrGetInSomeWayData();</tt><br>
   * <tt>     DataBlock block = transaction.transformToDataBlock(data);</tt><br>
   * <tt>     futureWrite = transaction.writeWhenPossible(block);</tt><br>
   * <br>
   * 5) Once you have finished, so this is the last block, you have to do the
   * following:<br>
   * If the last block is not empty:<br>
   * <tt>     DataBlock block = transaction.transformToDataBlock(data);</tt><br>
   * <tt>     block.setEOF(true);</tt><br>
   * Or if the last block is empty:<br>
   * <tt>     DataBlock block = transaction.transformToDataBlock(null);</tt><br>
   * Then <br>
   * <tt>     futureWrite = transaction.writeWhenPossible(block);</tt><br>
   * <tt>     futureWrite.awaitUninterruptibly();</tt><br>
   * <br>
   * 6) If everything is in success:<br>
   * <tt>     transaction.finalizeRequest();</tt><br>
   * <br>
   * And now wait for the transfer to finish:<br>
   * <tt>     futureReq.awaitUninterruptibly();</tt><br>
   * <tt>     R66Result result = futureReq.getResult();</tt><br>
   * <br>
   * 7) If there is the need to re-do, just re-execute the steps from 3 to 6.<br>
   * Don't forget at the very end to finish the global structure (steps 3 to 6 no
   * more executed):<br>
   * <tt>     networkTransaction.closeAll();</tt><br>
   * <br>
  * 8) In case of errors during steps 4 or 5 (and only those), call the
  * following:<br>
  * <tr>
  * transaction.transferInError(openR66Exception);
  * </tr>
  * <br>
  * <br>
  *
  * @see TestSendThroughClient Class as example of usage in test part
  */
 public abstract class SendThroughClient extends AbstractTransfer {
   protected final NetworkTransaction networkTransaction;
   protected LocalChannelReference localChannelReference;
   protected DbTaskRunner taskRunner;
 
   /**
    * @param future
    * @param remoteHost
    * @param filename
    * @param rulename
    * @param fileinfo
    * @param isMD5
    * @param blocksize
    * @param networkTransaction
    * @param id
    */
   protected SendThroughClient(final R66Future future, final String remoteHost,
                               final String filename, final String rulename,
                               final String fileinfo, final boolean isMD5,
                               final int blocksize, final long id,
                               final NetworkTransaction networkTransaction) {
     super(SendThroughClient.class, future, filename, rulename, fileinfo, isMD5,
           remoteHost, blocksize, id, null);
     this.networkTransaction = networkTransaction;
   }
 
   /**
    * DO NOT CALL THIS!
    */
   @Override
   public void run() {
     logger.error("DO NOT call this method for this class");
   }
 
   /**
    * Prior to call this method, the pipeline and NetworkTransaction must have
    * been initialized. It is the
    * responsibility of the caller to finish all network resources. Note that
    * this is only the first part of the
    * execution for this client.
    *
    * @return True if the initiate of the request is OK, else False
    */
   public boolean initiateRequest() {
     if (logger == null) {
       logger = WaarpLoggerFactory.getLogger(SendThroughClient.class);
     }
     final DbRule rule;
     try {
       rule = new DbRule(transferArgs.getRulename());
     } catch (final WaarpDatabaseException e) {
       logger.error("Cannot get Rule: " + transferArgs.getRulename() + ": {}",
                    e.getMessage());
       future.setResult(
           new R66Result(new OpenR66DatabaseGlobalException(e), null, true,
                         ErrorCode.Internal, null));
       future.setFailure(e);
       return false;
     }
     int mode = rule.getMode();
     if (transferArgs.isMD5()) {
       mode = RequestPacket.getModeMD5(mode);
     }
     final String sep =
         PartnerConfiguration.getSeparator(transferArgs.getRemoteHost());
     final RequestPacket request =
         new RequestPacket(transferArgs.getRulename(), mode,
                           transferArgs.getFilename(),
                           transferArgs.getBlockSize(), 0, transferArgs.getId(),
                           transferArgs.getTransferInfo(), -1, sep);
     // Not isRecv since it is the requester, so send => isSender is true
     final boolean isSender = true;
     try {
       try {
         // no starttime since immediate
         taskRunner = new DbTaskRunner(rule, isSender, request,
                                       transferArgs.getRemoteHost(), null);
       } catch (final WaarpDatabaseException e) {
         logger.error("Cannot get task: {}", e.getMessage());
         future.setResult(
             new R66Result(new OpenR66DatabaseGlobalException(e), null, true,
                           ErrorCode.Internal, null));
         future.setFailure(e);
         return false;
       }
       final ClientRunner runner =
           new ClientRunner(networkTransaction, taskRunner, future);
       runner.setSendThroughMode();
       OpenR66ProtocolNotYetConnectionException exc = null;
       for (int i = 0; i < Configuration.RETRYNB; i++) {
         try {
           localChannelReference = runner.initRequest();
           exc = null;
           break;
         } catch (final OpenR66ProtocolNoConnectionException e) {
           logger.error("Cannot Connect", e);
           future.setResult(
               new R66Result(e, null, true, ErrorCode.ConnectionImpossible,
                             taskRunner));
           finalizeInErrorTransferRequest(runner, taskRunner,
                                          ErrorCode.ConnectionImpossible);
           future.setFailure(e);
           return false;
         } catch (final OpenR66ProtocolPacketException e) {
           logger.error("Bad Protocol", e);
           future.setResult(new R66Result(e, null, true, ErrorCode.TransferError,
                                          taskRunner));
           future.setFailure(e);
           return false;
         } catch (final OpenR66ProtocolNotYetConnectionException e) {
           logger.debug("Not Yet Connected", e);
           exc = e;
         }
       }
       if (exc != null) {
         taskRunner.setLocalChannelReference(new LocalChannelReference());
         logger.error("Cannot Connect", exc);
         future.setResult(
             new R66Result(exc, null, true, ErrorCode.ConnectionImpossible,
                           taskRunner));
         future.setFailure(exc);
         return false;
       }
       try {
         localChannelReference.waitReadyForSendThrough();
       } catch (final OpenR66Exception e) {
         logger.error("Cannot Transfer", e);
         future.setResult(
             new R66Result(e, null, true, ErrorCode.Internal, taskRunner));
         future.setFailure(e);
         return false;
       }
       // now start the send from external data
       return true;
     } finally {
       if (taskRunner != null && (future.isFailed() || nolog)) {
         try {
           taskRunner.delete();
         } catch (final WaarpDatabaseException ignored) {
           // nothing
         }
       }
     }
   }
 
   /**
    * Finalize the request
    */
   public void finalizeRequest() {
     try {
       try {
         ChannelUtils.writeEndTransfer(localChannelReference);
       } catch (final OpenR66ProtocolPacketException e) {
         // An error occurs!
         try {
           localChannelReference.getSession().setFinalizeTransfer(false,
                                                                  new R66Result(
                                                                      e,
                                                                      localChannelReference.getSession(),
                                                                      false,
                                                                      ErrorCode.Internal,
                                                                      taskRunner));
         } catch (final OpenR66RunnerErrorException e1) {
           transferInError(e1);
           return;
         } catch (final OpenR66ProtocolSystemException e1) {
           transferInError(e1);
           return;
         }
       }
       localChannelReference.getFutureEndTransfer().awaitOrInterruptible();
       logger.debug("Await future End Transfer done: {}",
                    localChannelReference.getFutureEndTransfer().isSuccess());
       if (localChannelReference.getFutureEndTransfer().isSuccess()) {
         // send a validation
         localChannelReference.sessionNewState(R66FiniteDualStates.ENDREQUESTS);
         final EndRequestPacket validPacket =
             new EndRequestPacket(ErrorCode.CompleteOk.ordinal());
         final R66Session session = localChannelReference.getSession();
         if (session != null && session.getExtendedProtocol() &&
             session.getBusinessObject() != null &&
             session.getBusinessObject().getInfo(session) != null) {
           validPacket.setOptional(session.getBusinessObject().getInfo(session));
         }
         try {
           ChannelUtils.writeAbstractLocalPacket(localChannelReference,
                                                 validPacket, false);
         } catch (final OpenR66ProtocolPacketException ignored) {
           // nothing
         }
         if (!localChannelReference.getFutureRequest().awaitOrInterruptible()) {
           // valid it however
           localChannelReference.validateRequest(
               localChannelReference.getFutureEndTransfer().getResult());
         }
         if (taskRunner != null && taskRunner.isRequestOnRequested()) {
           localChannelReference.close();
         }
       } else {
         transferInError(null);
       }
     } finally {
       if (taskRunner != null) {
         if (future.isDone() && !future.isSuccess() || nolog) {
           try {
             taskRunner.delete();
           } catch (final WaarpDatabaseException ignored) {
             // nothing
           }
         }
       }
     }
   }
 
   /**
    * To be used in case of error after a correct initiate of the request
    *
    * @param e
    */
   public void transferInError(final OpenR66Exception e) {
     if (localChannelReference != null) {
       if (!localChannelReference.getFutureEndTransfer().getResult()
                                 .isAnswered()) {
         final R66Result result =
             new R66Result(e, localChannelReference.getSession(), true,
                           ErrorCode.TransferError, taskRunner);
         logger.error("Transfer in error", e);
         localChannelReference.sessionNewState(R66FiniteDualStates.ERROR);
         final ErrorPacket error = new ErrorPacket("Transfer in error",
                                                   ErrorCode.TransferError.getCode(),
                                                   ErrorPacket.FORWARDCLOSECODE);
         try {
           ChannelUtils.writeAbstractLocalPacket(localChannelReference, error,
                                                 false);
         } catch (final OpenR66ProtocolPacketException ignored) {
           // nothing
         }
         localChannelReference.invalidateRequest(result);
       }
       localChannelReference.close();
     }
   }
 
   /**
    * Write the next block when the channel is ready to prevent OOM
    *
    * @param block
    *
    * @return the ChannelFuture on the write operation
    *
    * @throws OpenR66ProtocolPacketException
    */
   public ChannelFuture writeWhenPossible(final DataBlock block)
       throws OpenR66ProtocolPacketException {
     return RetrieveRunner.writeWhenPossible(block, localChannelReference, null,
                                             null);
   }
 
   /**
    * Utility method for send through mode
    *
    * @param data the data byte, if null it is the last block
    * @param length length of data
    *
    * @return the DataBlock associated to the data
    */
   public DataBlock transformToDataBlock(final byte[] data, final int length) {
     final DataBlock block = new DataBlock();
     if (data == null) {
       // last block
       block.setEOF(true);
     } else {
       block.setBlock(data, length);
     }
     return block;
   }
 }