/*
 * 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.geode.distributed.internal.streaming;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.logging.log4j.Logger;

import org.apache.geode.CancelException;
import org.apache.geode.DataSerializer;
import org.apache.geode.GemFireRethrowable;
import org.apache.geode.InternalGemFireError;
import org.apache.geode.InternalGemFireException;
import org.apache.geode.SystemFailure;
import org.apache.geode.annotations.Immutable;
import org.apache.geode.cache.CacheClosedException;
import org.apache.geode.cache.query.internal.DefaultQuery;
import org.apache.geode.cache.query.internal.PRQueryTraceInfo;
import org.apache.geode.cache.query.internal.QueryMonitor;
import org.apache.geode.cache.query.internal.StructImpl;
import org.apache.geode.cache.query.internal.types.StructTypeImpl;
import org.apache.geode.cache.query.types.ObjectType;
import org.apache.geode.distributed.internal.ClusterDistributionManager;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.DistributionMessage;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.MessageWithReply;
import org.apache.geode.distributed.internal.PooledDistributionMessage;
import org.apache.geode.distributed.internal.ReplyException;
import org.apache.geode.distributed.internal.ReplyMessage;
import org.apache.geode.distributed.internal.ReplyProcessor21;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.HeapDataOutputStream;
import org.apache.geode.internal.InternalDataSerializer;
import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.PartitionedRegionQueryEvaluator;
import org.apache.geode.internal.cache.Token;
import org.apache.geode.internal.serialization.DeserializationContext;
import org.apache.geode.internal.serialization.SerializationContext;
import org.apache.geode.internal.serialization.Version;
import org.apache.geode.internal.util.BlobHelper;
import org.apache.geode.logging.internal.log4j.api.LogService;

/**
 * StreamingOperation is an abstraction for sending messages to multiple (or single) recipient
 * requesting a potentially large amount of data and receiving the reply with data chunked into
 * several messages.
 */
public abstract class StreamingOperation {
  private static final Logger logger = LogService.getLogger();

  /**
   * This is the number of bytes that need to be allowed in addition to data chunk to prevent
   * overflowing the socket buffer size in one message. For now, this is just an estimate
   */
  public static final int MSG_OVERHEAD = 200; // seems to need to be greater than 100

  public final InternalDistributedSystem sys;

  /**
   * Creates a new instance of StreamingOperation
   */
  public StreamingOperation(InternalDistributedSystem sys) {
    this.sys = sys;
  }

  /**
   * Returns normally if succeeded to get data, otherwise throws an exception
   *
   * @throws InterruptedException TODO-javadocs
   */
  public void getDataFromAll(Set recipients)
      throws org.apache.geode.cache.TimeoutException, InterruptedException {
    if (Thread.interrupted()) {
      throw new InterruptedException();
    }
    if (recipients.isEmpty()) {
      return;
    }

    StreamingProcessor processor = new StreamingProcessor(this.sys, recipients);
    DistributionMessage m = createRequestMessage(recipients, processor);
    this.sys.getDistributionManager().putOutgoing(m);
    // while() loop removed for bug 36983 - you can't loop on waitForReplies()
    try {
      // should we allow this to timeout?
      processor.waitForRepliesUninterruptibly();
    } catch (InternalGemFireException ex) {
      Throwable cause = ex.getCause();
      if (cause instanceof org.apache.geode.cache.TimeoutException) {
        throw (org.apache.geode.cache.TimeoutException) cause;
      }
      throw ex;
    } catch (ReplyException e) {
      e.handleCause();
      // throws exception
    }
  }

  /**
   * Override in subclass to instantiate request message
   */
  protected abstract DistributionMessage createRequestMessage(Set recipients,
      ReplyProcessor21 processor);

  /**
   * Called from separate thread when reply is processed.
   *
   * @return false if should abort (region was destroyed or cache was closed)
   */
  public boolean processChunk(List objects, InternalDistributedMember sender, int sequenceNum,
      boolean lastInSequence) {
    return processData(objects, sender, sequenceNum, lastInSequence);
  }

  /**
   * Override in subclass to do something useful with the data.
   *
   * @param sequenceNum the sequence of this data (0-based), in case ordering matters
   * @param lastInSequence true if this is the last chunk in the sequence
   * @return false to abort
   */

  protected abstract boolean processData(List objects, InternalDistributedMember sender,
      int sequenceNum, boolean lastInSequence);

  public class StreamingProcessor extends ReplyProcessor21 {
    protected volatile boolean abort = false;
    private final Map statusMap = new HashMap();

    protected final AtomicInteger msgsBeingProcessed = new AtomicInteger();

    class Status {
      int msgsProcessed = 0;
      int numMsgs = 0;

      /** Return true if this is the very last reply msg to process for this member */
      protected synchronized boolean trackMessage(StreamingReplyMessage m) {
        this.msgsProcessed++;

        if (m.lastMsg) {
          this.numMsgs = m.msgNum + 1;
        }
        if (logger.isDebugEnabled()) {
          logger.debug(
              "Streaming Message Tracking Status: Processor id: {}; Sender: {}; Messages Processed: {}; NumMsgs: {}",
              getProcessorId(), m.getSender(), this.msgsProcessed, this.numMsgs);
        }

        // this.numMsgs starts out as zero and gets initialized
        // only when we get a lastMsg true.
        // Since we increment msgsProcessed, the following condition
        // cannot be true until sometime after we've received the
        // lastMsg, and signals that all messages have been processed
        return this.msgsProcessed == this.numMsgs;
      }
    }


    public StreamingProcessor(final InternalDistributedSystem system,
        InternalDistributedMember member) {
      super(system, member);
    }

    public StreamingProcessor(InternalDistributedSystem system, Set members) {
      super(system, members);
    }

    @Override
    public void process(DistributionMessage msg) {
      // ignore messages from members not in the wait list
      if (!waitingOnMember(msg.getSender())) {
        return;
      }
      this.msgsBeingProcessed.incrementAndGet();
      try {
        StreamingReplyMessage m = (StreamingReplyMessage) msg;
        boolean isLast = true; // is last message for this member?
        List objects = m.getObjects();
        if (objects != null) { // CONSTRAINT: objects should only be null if there's no data at all
          // Bug 37461: don't allow abort to be reset.
          boolean isAborted = this.abort; // volatile fetch
          if (!isAborted) {
            isAborted = !processChunk(objects, m.getSender(), m.msgNum, m.lastMsg);
            if (isAborted) {
              this.abort = true; // volatile store
            }
          }
          isLast = isAborted || trackMessage(m); // interpret msgNum
          // @todo ezoerner send an abort message to data provider if
          // !doContinue (region was destroyed or cache closed);
          // also provide ability to explicitly cancel
        } else {
          // if a null chunk was received (no data), then
          // we're done with that member
          isLast = true;
        }
        if (isLast) {
          super.process(msg, false); // removes from members and cause us to
                                     // ignore future messages received from that member
        }
      } finally {
        this.msgsBeingProcessed.decrementAndGet();
        checkIfDone(); // check to see if decrementing msgsBeingProcessed requires signalling to
                       // proceed
      }
    }

    /**
     * Contract of {@link ReplyProcessor21#stillWaiting()} is to never return true after returning
     * false.
     */
    private volatile boolean finishedWaiting = false;

    /**
     * Overridden to wait for messages being currently processed: This situation can come about if a
     * member departs while we are still processing data from that member
     */
    @Override
    protected boolean stillWaiting() {
      if (finishedWaiting) { // volatile fetch
        return false;
      }
      if (this.msgsBeingProcessed.get() > 0) {
        // to fix bug 37391 always wait for msgsBeingPRocessod to go to 0,
        // even if abort is true
        return true;
      }
      // volatile fetches
      finishedWaiting = finishedWaiting || this.abort || !super.stillWaiting();
      return !finishedWaiting;
    }


    @Override
    public String toString() {
      return "<" + this.getClass().getName() + " " + this.getProcessorId() + " waiting for "
          + numMembers() + " replies" + (exception == null ? "" : (" exception: " + exception))
          + " from " + membersToString() + "; waiting for " + this.msgsBeingProcessed.get()
          + " messages in the process of being processed" + ">";
    }

    protected boolean trackMessage(StreamingReplyMessage m) {
      Status status;
      synchronized (this) {
        status = (Status) this.statusMap.get(m.getSender());
        if (status == null) {
          status = new Status();
          this.statusMap.put(m.getSender(), status);
        }
      }
      return status.trackMessage(m);
    }

  }

  public abstract static class RequestStreamingMessage extends PooledDistributionMessage
      implements MessageWithReply {

    protected int processorId;

    @Override
    public int getProcessorId() {
      return this.processorId;
    }

    @Override
    protected void process(final ClusterDistributionManager dm) {
      Throwable thr = null;
      ReplyException rex = null;
      Object nextObject = null;
      Object failedObject = null;
      int socketBufferSize = dm.getSystem().getConfig().getSocketBufferSize();
      int chunkSize = socketBufferSize - MSG_OVERHEAD;
      HeapDataOutputStream outStream =
          new HeapDataOutputStream(chunkSize, getSender().getVersionObject());
      boolean sentFinalMessage = false;
      boolean receiverCacheClosed = false;
      int msgNum = 0;

      try {
        do {
          int numObjectsInChunk = 0;
          // boolean firstObject = true;

          // always write at least one object, allowing expansion
          // if we have an object already that didn't get added, then use
          // that object instead of getting another one
          if (failedObject == null) {
            nextObject = getNextReplyObject();
          } else {
            nextObject = failedObject;
            failedObject = null;
          }

          if (nextObject != Token.END_OF_STREAM) {
            numObjectsInChunk = 1;
            BlobHelper.serializeTo(nextObject, outStream);

            // for the next objects, disallow stream from allocating more storage
            do {
              outStream.disallowExpansion(CHUNK_FULL); // sets the mark where rollback occurs on
                                                       // CHUNK_FULL

              nextObject = getNextReplyObject();

              if (nextObject != Token.END_OF_STREAM) {
                try {
                  BlobHelper.serializeTo(nextObject, outStream);
                  numObjectsInChunk++;
                } catch (GemFireRethrowable e) {
                  // can only be thrown when expansion is disallowed
                  // and buffer is automatically reset to point where it was disallowed
                  failedObject = nextObject;
                  break;
                }
              }
            } while (nextObject != Token.END_OF_STREAM);
          }

          try {
            replyWithData(dm, outStream, numObjectsInChunk, msgNum++,
                nextObject == Token.END_OF_STREAM);
            if (nextObject == Token.END_OF_STREAM) {
              sentFinalMessage = true;
            }
          } catch (CancelException e) {
            receiverCacheClosed = true;
            break; // receiver no longer cares
          }
          outStream.reset(); // ready for reuse, assumes replyWithData
                             // does not queue the message but outStream has
                             // already been used
        } while (nextObject != Token.END_OF_STREAM);
        // } catch (CancelException e) {
        // // if cache is closed, we cannot send a reply (correct?)
      } catch (VirtualMachineError err) {
        SystemFailure.initiateFailure(err);
        // If this ever returns, rethrow the error. We're poisoned
        // now, so don't let this thread continue.
        throw err;
      } catch (Throwable t) {
        // Whenever you catch Error or Throwable, you must also
        // catch VirtualMachineError (see above). However, there is
        // _still_ a possibility that you are dealing with a cascading
        // error condition, so you also need to check to see if the JVM
        // is still usable:
        SystemFailure.checkFailure();
        thr = t;
      }

      if (thr != null) {
        rex = new ReplyException(thr);
        replyWithException(dm, rex);
      } else if (!sentFinalMessage && !receiverCacheClosed) {
        throw new InternalGemFireError(
            "this should not happen");
        // replyNoData(dm);
      }
    }

    /**
     * override in subclass to provide reply data. terminate by returning Token.END_OF_STREAM
     */
    protected abstract Object getNextReplyObject() throws InterruptedException;

    // private void replyNoData(DistributionManager dm) {
    // StreamingReplyMessage.send(getSender(), this.processorId, null, dm, null, 0, 0, true);
    // }

    protected void replyWithData(ClusterDistributionManager dm, HeapDataOutputStream outStream,
        int numObjects, int msgNum, boolean lastMsg) {
      StreamingReplyMessage.send(getSender(), this.processorId, null, dm, outStream, numObjects,
          msgNum, lastMsg);
    }

    protected void replyWithException(ClusterDistributionManager dm, ReplyException rex) {
      StreamingReplyMessage.send(getSender(), this.processorId, rex, dm, null, 0, 0, true);
    }

    @Override
    public void fromData(DataInput in,
        DeserializationContext context) throws IOException, ClassNotFoundException {
      super.fromData(in, context);
      this.processorId = in.readInt();
    }

    @Override
    public void toData(DataOutput out,
        SerializationContext context) throws IOException {
      super.toData(out, context);
      out.writeInt(this.processorId);
    }

    @Override
    public String toString() {
      StringBuffer buff = new StringBuffer();
      buff.append(getClass().getName());
      buff.append("'; sender=");
      buff.append(getSender());
      buff.append("; processorId=");
      buff.append(this.processorId);
      buff.append(")");
      return buff.toString();
    }
  }

  public static class StreamingReplyMessage extends ReplyMessage {

    /** the number of this message */
    protected int msgNum;

    /** whether this message is the last one in this series */
    protected boolean lastMsg;

    private transient HeapDataOutputStream chunkStream; // used only on sending side, null means
                                                        // abort
    private transient int numObjects; // used only on sending side
    private transient List objectList = null; // used only on receiving side

    private boolean pdxReadSerialized = false; // used to read PDX types in serialized form.
    private transient boolean isCanceled = false; // used only on receiving side and if
                                                  // messageProcessor is of type
                                                  // PartitionedRegionQueryEvaluator.StreamingQueryPartitionResponse


    /**
     * @param chunkStream the data to send back, if null then all the following parameters are
     *        ignored and any future replies from this member will be ignored, and the streaming of
     *        chunks is considered aborted by the receiver.
     *
     * @param msgNum message number in this series (0-based)
     * @param lastMsg if this is the last message in this series
     */
    public static void send(InternalDistributedMember recipient, int processorId,
        ReplyException exception, DistributionManager dm, HeapDataOutputStream chunkStream,
        int numObjects, int msgNum, boolean lastMsg) {
      send(recipient, processorId, exception, dm, chunkStream, numObjects, msgNum, lastMsg, false);
    }

    public static void send(InternalDistributedMember recipient, int processorId,
        ReplyException exception, DistributionManager dm, HeapDataOutputStream chunkStream,
        int numObjects, int msgNum, boolean lastMsg, boolean pdxReadSerialized) {
      StreamingReplyMessage replyMessage = new StreamingReplyMessage();
      replyMessage.processorId = processorId;

      if (exception != null) {
        replyMessage.setException(exception);
        logger.debug("Replying with exception: {}", replyMessage, exception);
      }

      replyMessage.chunkStream = chunkStream;
      replyMessage.numObjects = numObjects;
      replyMessage.setRecipient(recipient);
      replyMessage.msgNum = msgNum;
      replyMessage.lastMsg = lastMsg;
      replyMessage.pdxReadSerialized = pdxReadSerialized;
      dm.putOutgoing(replyMessage);
    }

    public int getMessageNumber() {
      return this.msgNum;
    }

    public boolean isLastMessage() {
      return this.lastMsg;
    }

    public boolean isCanceled() {
      return isCanceled;
    }

    /** Return the objects in this chunk as a List, used only on receiving side */
    public List getObjects() {
      return this.objectList;
    }

    @Override
    public int getDSFID() {
      return STREAMING_REPLY_MESSAGE;
    }


    @Override
    public void fromData(DataInput in,
        DeserializationContext context) throws IOException, ClassNotFoundException {
      int n;
      super.fromData(in, context);
      n = in.readInt();
      this.msgNum = in.readInt();
      this.lastMsg = in.readBoolean();
      this.pdxReadSerialized = in.readBoolean();
      Version senderVersion = InternalDataSerializer.getVersionForDataStream(in);
      boolean isSenderAbove_8_1 = senderVersion.compareTo(Version.GFE_81) > 0;
      InternalCache cache = null;
      Boolean initialPdxReadSerialized = false;
      try {
        cache =
            (InternalCache) GemFireCacheImpl.getForPdx("fromData invocation in StreamingOperation");
        initialPdxReadSerialized = cache.getPdxReadSerializedOverride();
      } catch (CacheClosedException e) {
        logger.debug("Cache is closed. PdxReadSerializedOverride set to false");
      }
      if (n == -1) {
        this.objectList = null;
      } else {
        this.numObjects = n; // for benefit of toString()
        this.objectList = new ArrayList(n);
        // Check if the PDX types needs to be kept in serialized form.
        // This will make readObject() to return PdxInstance form.
        if (this.pdxReadSerialized && cache != null) {
          cache.setPdxReadSerializedOverride(true);
        }
        try {
          ReplyProcessor21 messageProcessor = ReplyProcessor21.getProcessor(processorId);
          boolean isQueryMessageProcessor =
              messageProcessor instanceof PartitionedRegionQueryEvaluator.StreamingQueryPartitionResponse;
          ObjectType elementType = null;
          if (isQueryMessageProcessor) {
            elementType =
                ((PartitionedRegionQueryEvaluator.StreamingQueryPartitionResponse) messageProcessor)
                    .getResultType();
          }

          boolean lowMemoryDetected = false;
          for (int i = 0; i < n; i++) {
            // TestHook used in ResourceManagerWithQueryMonitorDUnitTest.
            // will simulate an critical memory event after a certain number of calls to
            // doTestHook(BEFORE_ADD_OR_UPDATE_MAPPING_OR_DESERIALIZING_NTH_STREAMINGOPERATION)
            if (DefaultQuery.testHook != null) {
              DefaultQuery.testHook.doTestHook(
                  DefaultQuery.TestHook.SPOTS.BEFORE_ADD_OR_UPDATE_MAPPING_OR_DESERIALIZING_NTH_STREAMINGOPERATION,
                  null, null);
            }
            if (isQueryMessageProcessor && QueryMonitor.isLowMemory()) {
              lowMemoryDetected = true;
              break;
            }
            Object o = DataSerializer.readObject(in);
            if (isQueryMessageProcessor && elementType != null && elementType.isStructType()) {
              boolean convertToStruct = isSenderAbove_8_1;
              if (convertToStruct && i == 0) {
                convertToStruct = !(o instanceof PRQueryTraceInfo);
              }
              if (convertToStruct) {
                o = new StructImpl((StructTypeImpl) elementType, (Object[]) o);
              }
            }
            this.objectList.add(o);
          }
          if (lowMemoryDetected) {
            isCanceled = true;
            // TestHook to help verify that objects have been rejected.
            if (DefaultQuery.testHook != null) {
              DefaultQuery.testHook.doTestHook(
                  DefaultQuery.TestHook.SPOTS.LOW_MEMORY_WHEN_DESERIALIZING_STREAMINGOPERATION,
                  null, null);
            }
          }
        } finally {
          if (this.pdxReadSerialized && cache != null) {
            cache.setPdxReadSerializedOverride(initialPdxReadSerialized);
          }
        }
      }
    }

    @Override
    public void toData(DataOutput out,
        SerializationContext context) throws IOException {
      super.toData(out, context);
      if (this.chunkStream == null) {
        out.writeInt(-1);
      } else {
        out.writeInt(this.numObjects);
      }
      out.writeInt(this.msgNum);
      out.writeBoolean(this.lastMsg);
      out.writeBoolean(this.pdxReadSerialized);
      if (this.chunkStream != null && this.numObjects > 0) {
        this.chunkStream.sendTo(out);
      }
    }

    @Override
    public String toString() {
      StringBuffer buff = new StringBuffer();
      buff.append(getClass().getName());
      buff.append("(processorId=");
      buff.append(this.processorId);
      buff.append(" from ");
      buff.append(this.getSender());
      ReplyException ex = this.getException();
      if (ex != null) {
        buff.append(" with exception ");
        buff.append(ex);
      }
      buff.append(";numObjects=");
      buff.append(this.numObjects);
      buff.append(";msgNum ");
      buff.append(this.msgNum);
      buff.append(";lastMsg=");
      buff.append(this.lastMsg);
      if (this.objectList != null) {
        buff.append(";objectList(size=");
        buff.append(this.objectList.size());
        buff.append(")");
      } else {
        buff.append(";chunkStream=");
        buff.append(this.chunkStream);
      }
      buff.append(")");
      return buff.toString();
    }

  }

  @Immutable
  public static final GemFireRethrowable CHUNK_FULL = new GemFireRethrowable();
}