/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * license agreements; and to You under the Apache License, version 2.0:
 *
 *   https://www.apache.org/licenses/LICENSE-2.0
 *
 * This file is part of the Apache Pekko project, which was derived from Akka.
 */

/*
 * Copyright (C) since 2016 Lightbend Inc. <https://www.lightbend.com>
 */

package org.apache.pekko.stream.connectors.couchbase.javadsl

import org.apache.pekko
import pekko.NotUsed
import pekko.stream.connectors.couchbase.{
  scaladsl,
  CouchbaseDeleteResult,
  CouchbaseSessionSettings,
  CouchbaseWriteResult,
  CouchbaseWriteSettings
}
import pekko.stream.javadsl.Flow
import com.couchbase.client.java.document.{ Document, JsonDocument }

/**
 * Java API: Factory methods for Couchbase flows.
 */
object CouchbaseFlow {

  /**
   * Create a flow to query Couchbase for by `id` and emit [[com.couchbase.client.java.document.JsonDocument JsonDocument]]s.
   */
  def fromId(sessionSettings: CouchbaseSessionSettings, bucketName: String): Flow[String, JsonDocument, NotUsed] =
    scaladsl.CouchbaseFlow.fromId(sessionSettings, bucketName).asJava

  /**
   * Create a flow to query Couchbase for by `id` and emit documents of the given class.
   */
  def fromId[T <: Document[?]](sessionSettings: CouchbaseSessionSettings,
      bucketName: String,
      target: Class[T]): Flow[String, T, NotUsed] =
    scaladsl.CouchbaseFlow.fromId(sessionSettings, bucketName, target).asJava

  /**
   * Create a flow to update or insert a Couchbase [[com.couchbase.client.java.document.JsonDocument JsonDocument]].
   */
  def upsert(sessionSettings: CouchbaseSessionSettings,
      writeSettings: CouchbaseWriteSettings,
      bucketName: String): Flow[JsonDocument, JsonDocument, NotUsed] =
    scaladsl.CouchbaseFlow.upsert(sessionSettings, writeSettings, bucketName).asJava

  /**
   * Create a flow to update or insert a Couchbase document of the given class.
   */
  def upsertDoc[T <: Document[?]](sessionSettings: CouchbaseSessionSettings,
      writeSettings: CouchbaseWriteSettings,
      bucketName: String): Flow[T, T, NotUsed] =
    scaladsl.CouchbaseFlow.upsertDoc(sessionSettings, writeSettings, bucketName).asJava

  /**
   * Create a flow to update or insert a Couchbase document of the given class and emit a result so that write failures
   * can be handled in-stream.
   */
  def upsertDocWithResult[T <: Document[?]](sessionSettings: CouchbaseSessionSettings,
      writeSettings: CouchbaseWriteSettings,
      bucketName: String): Flow[T, CouchbaseWriteResult[T], NotUsed] =
    scaladsl.CouchbaseFlow.upsertDocWithResult(sessionSettings, writeSettings, bucketName).asJava

  /**
   * Create a flow to replace a Couchbase [[com.couchbase.client.java.document.JsonDocument JsonDocument]].
   */
  def replace(sessionSettings: CouchbaseSessionSettings,
      writeSettings: CouchbaseWriteSettings,
      bucketName: String): Flow[JsonDocument, JsonDocument, NotUsed] =
    scaladsl.CouchbaseFlow.replace(sessionSettings, writeSettings, bucketName).asJava

  /**
   * Create a flow to replace a Couchbase document of the given class.
   */
  def replaceDoc[T <: Document[?]](sessionSettings: CouchbaseSessionSettings,
      writeSettings: CouchbaseWriteSettings,
      bucketName: String): Flow[T, T, NotUsed] =
    scaladsl.CouchbaseFlow.replaceDoc(sessionSettings, writeSettings, bucketName).asJava

  /**
   * Create a flow to replace a Couchbase document of the given class and emit a result so that write failures
   * can be handled in-stream.
   */
  def replaceDocWithResult[T <: Document[?]](sessionSettings: CouchbaseSessionSettings,
      writeSettings: CouchbaseWriteSettings,
      bucketName: String): Flow[T, CouchbaseWriteResult[T], NotUsed] =
    scaladsl.CouchbaseFlow.replaceDocWithResult(sessionSettings, writeSettings, bucketName).asJava

  /**
   * Create a flow to delete documents from Couchbase by `id`. Emits the same `id`.
   */
  def delete(sessionSettings: CouchbaseSessionSettings,
      writeSettings: CouchbaseWriteSettings,
      bucketName: String): Flow[String, String, NotUsed] =
    scaladsl.CouchbaseFlow.delete(sessionSettings, writeSettings, bucketName).asJava

  /**
   * Create a flow to delete documents from Couchbase by `id` and emit operation outcome containing the same `id`.
   */
  def deleteWithResult(sessionSettings: CouchbaseSessionSettings,
      writeSettings: CouchbaseWriteSettings,
      bucketName: String): Flow[String, CouchbaseDeleteResult, NotUsed] =
    scaladsl.CouchbaseFlow.deleteWithResult(sessionSettings, writeSettings, bucketName).asJava

}
