code: /*
   * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
   */
  
  package io.airbyte.integrations.destination.dest_state_lifecycle_manager;
  
  import com.google.common.annotations.VisibleForTesting;
  import io.airbyte.protocol.models.AirbyteMessage;
  import java.util.Collections;
  import java.util.LinkedList;
  import java.util.List;
  import java.util.Queue;
  
  /**
   * This {@link DestStateLifecycleManager} handles any state where there is a guarantee that any
   * single state message represents the state for the ENTIRE connection. At the time of writing,
   * GLOBAL and LEGACY state types are the state type that match this pattern.
   *
   * Does NOT store duplicates. Because each state message represents the entire state for the
   * connection, it only stores (and emits) the LAST state it received at each phase.
   */
  public class DestSingleStateLifecycleManager implements DestStateLifecycleManager {
  
    private AirbyteMessage lastPendingState;
    private AirbyteMessage lastFlushedState;
    private AirbyteMessage lastCommittedState;
  
    @Override
    public void addState(final AirbyteMessage message) {
      lastPendingState = message;
    }
  
    @VisibleForTesting
    Queue<AirbyteMessage> listPending() {
      return stateMessageToQueue(lastPendingState);
    }
  
    @Override
    public void markPendingAsFlushed() {
      if (lastPendingState != null) {
        lastFlushedState = lastPendingState;
        lastPendingState = null;
      }
    }
  
    @Override
    public Queue<AirbyteMessage> listFlushed() {
      return stateMessageToQueue(lastFlushedState);
    }
  
    @Override
    public void markFlushedAsCommitted() {
      if (lastFlushedState != null) {
        lastCommittedState = lastFlushedState;
        lastFlushedState = null;
      }
    }
  
    @Override
    public Queue<AirbyteMessage> listCommitted() {
      return stateMessageToQueue(lastCommittedState);
    }
  
    private static Queue<AirbyteMessage> stateMessageToQueue(final AirbyteMessage stateMessage) {
      return new LinkedList<>(stateMessage == null ? Collections.emptyList() : List.of(stateMessage));
    }
  
  }
  