code: /*
   * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
   */
  
  package io.airbyte.integrations.base;
  
  import io.airbyte.commons.concurrency.VoidCallable;
  import io.airbyte.commons.functional.CheckedConsumer;
  import io.airbyte.protocol.models.AirbyteMessage;
  
  /**
   * Interface for the destination's consumption of incoming records wrapped in an
   * {@link io.airbyte.protocol.models.AirbyteMessage}.
   *
   * This is via the accept method, which commonly handles parsing, validation, batching and writing
   * of the transformed data to the final destination i.e. the technical system data is being written
   * to.
   *
   * Lifecycle:
   * <ul>
   * <li>1. Instantiate consumer.</li>
   * <li>2. start() to initialize any resources that need to be created BEFORE the consumer consumes
   * any messages.</li>
   * <li>3. Consumes ALL records via {@link AirbyteMessageConsumer#accept(AirbyteMessage)}</li>
   * <li>4. Always (on success or failure) finalize by calling
   * {@link AirbyteMessageConsumer#close()}</li>
   * </ul>
   * We encourage implementing this interface using the {@link FailureTrackingAirbyteMessageConsumer}
   * class.
   */
  public interface AirbyteMessageConsumer extends CheckedConsumer<AirbyteMessage, Exception>, AutoCloseable {
  
    void start() throws Exception;
  
    @Override
    void accept(AirbyteMessage message) throws Exception;
  
    @Override
    void close() throws Exception;
  
    /**
     * Append a function to be called on {@link AirbyteMessageConsumer#close}.
     */
    static AirbyteMessageConsumer appendOnClose(final AirbyteMessageConsumer consumer, final VoidCallable voidCallable) {
      return new AirbyteMessageConsumer() {
  
        @Override
        public void start() throws Exception {
          consumer.start();
        }
  
        @Override
        public void accept(final AirbyteMessage message) throws Exception {
          consumer.accept(message);
        }
  
        @Override
        public void close() throws Exception {
          consumer.close();
          voidCallable.call();
        }
  
      };
    }
  
  }
  