//设置package包名称以及导入依赖的类
package time.impl

import time.api.TimeService
import com.lightbend.lagom.javadsl.api.ServiceCall
import akka.NotUsed
import java.util.concurrent.CompletableFuture
import java.util.concurrent.CompletionStage
import java.time.LocalTime
import java.time.ZoneId

class TimeServiceImpl extends TimeService{

  
  override def timeAt(tz: String): ServiceCall[NotUsed, String] = {
    new ServiceCall[NotUsed, String] {
      
      override def invoke(obj: NotUsed) : CompletionStage[String] = {
        val c = new CompletableFuture[String]
        c.complete(LocalTime.now(ZoneId.of(tz, ZoneId.SHORT_IDS)).toString)
        c
      }
      
    }
  }
  
} 

//设置package包名称以及导入依赖的类
package akka.stream.alpakka.ironmq

import java.util.concurrent.CompletionStage

import akka.Done
import akka.stream.alpakka.ironmq.scaladsl.{
  Committable => ScalaCommittable,
  CommittableMessage => ScalaCommittableMessage
}

import scala.compat.java8.FutureConverters
import scala.concurrent.Future


package object javadsl {

  import FutureConverters._

  private[javadsl] implicit class RichScalaCommittableMessage(cm: ScalaCommittableMessage) {
    def asJava: CommittableMessage = new CommittableMessage {
      override def message: Message = cm.message
      override def commit(): CompletionStage[Done] = cm.commit().toJava
    }
  }

  private[javadsl] implicit class RichScalaCommittable(cm: ScalaCommittable) {
    def asJava: Committable = new Committable {
      override def commit(): CompletionStage[Done] = cm.commit().toJava
    }
  }

  private[javadsl] implicit class RichCommittableMessage(cm: CommittableMessage) {
    def asScala: ScalaCommittableMessage = new ScalaCommittableMessage {
      override def message: Message = cm.message
      override def commit(): Future[Done] = cm.commit().toScala
    }
  }

  private[javadsl] implicit class RichCommittable(cm: Committable) {
    def asScala: ScalaCommittable = new ScalaCommittable {
      override def commit(): Future[Done] = cm.commit().toScala
    }
  }

} 

//设置package包名称以及导入依赖的类
package akka.stream.alpakka.cassandra.javadsl

import java.util.concurrent.CompletionStage
import java.util.function.BiFunction

import akka.Done
import akka.stream.javadsl.Sink
import com.datastax.driver.core.{BoundStatement, PreparedStatement, Session}
import akka.stream.alpakka.cassandra.scaladsl.{CassandraSink => ScalaCSink}

import scala.compat.java8.FutureConverters._
import scala.concurrent.ExecutionContext

object CassandraSink {

  def create[T](parallelism: Int,
                statement: PreparedStatement,
                statementBinder: BiFunction[T, PreparedStatement, BoundStatement],
                session: Session,
                executionContext: ExecutionContext): Sink[T, CompletionStage[Done]] = {
    val sink =
      ScalaCSink.apply[T](parallelism, statement, (t, p) => statementBinder.apply(t, p))(session, executionContext)

    sink.mapMaterializedValue(_.toJava).asJava
  }

} 

//设置package包名称以及导入依赖的类
package akka.stream.alpakka.azure.storagequeue.javadsl

import com.microsoft.azure.storage.queue.{CloudQueue, CloudQueueMessage}
import akka.stream.alpakka.azure.storagequeue.{
  AzureQueueSinkFunctions,
  Delete,
  DeleteOrUpdateMessage,
  UpdateVisibility
}
import akka.stream.javadsl.Sink
import akka.Done
import scala.concurrent.Future
import java.util.concurrent.CompletionStage
import java.util.function.Supplier

object AzureQueueSink {

  
  def create(cloudQueue: Supplier[CloudQueue]): Sink[MessageAndDeleteOrUpdate, CompletionStage[Done]] =
    AzureQueueSink.fromFunction[MessageAndDeleteOrUpdate](
      input => AzureQueueSinkFunctions.deleteOrUpdateMessage(() => cloudQueue.get)(input.message, input.op)
    )
} 

//设置package包名称以及导入依赖的类
package akka.stream.alpakka.sse
package javadsl

import akka.NotUsed
import akka.http.javadsl.model.{HttpRequest, HttpResponse, Uri}
import akka.http.scaladsl.model.{HttpResponse => SHttpResponse}
import akka.stream.Materializer
import akka.stream.javadsl.Source
import akka.http.javadsl.model.sse.ServerSentEvent
import java.util.Optional
import java.util.concurrent.CompletionStage
import java.util.function.{Function => JFunction}
import scala.compat.java8.FutureConverters
import scala.compat.java8.OptionConverters


  def create(uri: Uri,
             send: JFunction[HttpRequest, CompletionStage[HttpResponse]],
             lastEventId: Optional[String],
             mat: Materializer): Source[ServerSentEvent, NotUsed] = {
    val eventSource =
      scaladsl.EventSource(
        uri.asInstanceOf[akka.http.impl.model.JavaUri].uri,
        send(_).toScala.map(_.asInstanceOf[SHttpResponse])(mat.executionContext),
        lastEventId.asScala
      )(mat)
    eventSource.asJava
  }
} 

//设置package包名称以及导入依赖的类
package io.scalac.lagom.utils

import java.util.concurrent.CompletableFuture._
import java.util.concurrent.CompletionStage

import com.lightbend.lagom.javadsl.api.ServiceCall

object Implicits {
  implicit def requestToServiceCallWithCompletedFuture[Req, Resp](reqFunc: Req => Resp): ServiceCall[Req, Resp] = {
    new ServiceCall[Req, Resp] {
      override def invoke(request: Req): CompletionStage[Resp] = {
        completedFuture(reqFunc(request))
      }
    }
  }

  implicit def requestToServiceCall[Req, Resp](reqFunc: Req => CompletionStage[Resp]): ServiceCall[Req, Resp] = {
    new ServiceCall[Req, Resp] {
      override def invoke(request: Req): CompletionStage[Resp] = {
        reqFunc(request)
      }
    }
  }

  implicit def sFun1ToAkkaJapiFun[T, R](sFun1: T => R): akka.japi.function.Function[T, R] =
    new akka.japi.function.Function[T, R] {
      @scala.throws[Exception](classOf[Exception])
      override def apply(param: T): R = sFun1.apply(param)
    }

  implicit def sFunToAkkaEffect(sFun: () => Unit): akka.japi.Effect =
    new akka.japi.Effect {
      @scala.throws[Exception](classOf[Exception])
      override def apply(): Unit = sFun()
    }

  implicit def asJavaBiFunction[T, U, R](sFun: (T, U) => R): java.util.function.BiFunction[T, U, R] =
    new java.util.function.BiFunction[T, U, R] {
      override def apply(t: T, u: U): R = sFun(t, u)
    }

  implicit def asJavaConsumer[T](sFun: T => Unit): java.util.function.Consumer[T] =
    new java.util.function.Consumer[T] {
      override def accept(t: T): Unit = sFun(t)
    }

  implicit def asJavaFunction[T, R](sFun: T => R): java.util.function.Function[T, R] =
    new java.util.function.Function[T, R] {
      override def apply(t: T): R = sFun(t)
    }

  implicit def asJavaBiConsumer[T, U](sFun: (T, U) => Unit): java.util.function.BiConsumer[T, U] =
    new java.util.function.BiConsumer[T, U] {
      override def accept(t: T, u: U): Unit = sFun(t, u)
    }
} 

//设置package包名称以及导入依赖的类
package play.socketio.javadsl

import java.util.Optional
import java.util.concurrent.CompletionStage
import java.util.function.{ BiFunction, Function }

import akka.NotUsed
import akka.stream.Materializer
import akka.stream.javadsl.Flow
import com.fasterxml.jackson.databind.JsonNode
import play.api.libs.json.Json
import play.mvc.Http.RequestHeader
import play.socketio.{ SocketIOConfig, SocketIOEvent, SocketIOSession, SocketIOSessionFlow }

import scala.concurrent.ExecutionContext
import scala.Function.unlift
import scala.compat.java8.FutureConverters._
import scala.compat.java8.OptionConverters._


private[javadsl] object SocketIOSessionFlowHelper {

  def createEngineIOSessionHandler[SessionData](
    config:                     SocketIOConfig,
    connectCallback:            BiFunction[RequestHeader, String, CompletionStage[SessionData]],
    errorHandler:               Function[Throwable, Optional[JsonNode]],
    defaultNamespaceCallback:   Function[SocketIOSession[SessionData], Flow[SocketIOEvent, SocketIOEvent, NotUsed]],
    connectToNamespaceCallback: BiFunction[SocketIOSession[SessionData], String, Optional[Flow[SocketIOEvent, SocketIOEvent, NotUsed]]]
  )(implicit ec: ExecutionContext, mat: Materializer) = {
    SocketIOSessionFlow.createEngineIOSessionHandler[SessionData](
      config,
      (request, sid) => connectCallback(request.asJava, sid).toScala,
      unlift(t => errorHandler(t).asScala.map(Json.toJson(_))),
      session => defaultNamespaceCallback(session).asScala,
      unlift {
        case (session, sid) => connectToNamespaceCallback(session, sid).asScala.map(_.asScala)
      }
    )
  }
} 

//设置package包名称以及导入依赖的类
package com.lightbend.lagom.javadsl.persistence

import java.util.concurrent.CompletionStage
import java.util.function.BiConsumer

import akka.actor.ActorRef

package object testkit {

  implicit class pipe[T](val stage: CompletionStage[T]) extends AnyVal {
    def pipeTo(recipient: ActorRef): Unit = {
      stage.whenComplete(new BiConsumer[T, Throwable] {
        override def accept(value: T, e: Throwable): Unit = {
          if (value != null) recipient ! value
          if (e != null) recipient ! e
        }
      })
    }
  }
} 

//设置package包名称以及导入依赖的类
package com.lightbend.lagom.internal.testkit

import java.net.URI
import java.util.Optional
import java.util.concurrent.CompletionStage

import scala.compat.java8.FutureConverters._
import scala.concurrent.ExecutionContext
import scala.concurrent.Future
import com.lightbend.lagom.javadsl.api.Descriptor
import javax.inject.Inject
import javax.inject.Singleton

import com.lightbend.lagom.javadsl.client.{ CircuitBreakersPanel, CircuitBreakingServiceLocator }

@Singleton
private[lagom] class TestServiceLocator @Inject() (
  circuitBreakers: CircuitBreakersPanel,
  port:            TestServiceLocatorPort,
  implicit val ec: ExecutionContext
) extends CircuitBreakingServiceLocator(circuitBreakers) {

  private val futureUri = port.port.map(p => URI.create("http://localhost:" + p))

  override def locate(name: String, call: Descriptor.Call[_, _]): CompletionStage[Optional[URI]] =
    futureUri.map(uri => Optional.of(uri)).toJava
}

private[lagom] final case class TestServiceLocatorPort(port: Future[Int]) 

//设置package包名称以及导入依赖的类
package com.lightbend.lagom.internal.javadsl.testkit

import java.util.concurrent.CompletionStage

import akka.Done
import akka.actor.ActorRef
import akka.stream.Materializer
import akka.stream.javadsl.{ Flow, Source }
import com.lightbend.lagom.internal.testkit.InternalSubscriberStub
import com.lightbend.lagom.javadsl.api.broker.{ Subscriber, Topic }

import scala.compat.java8.FutureConverters.toJava

private[lagom] class TopicStub[T](val topicId: Topic.TopicId, topicBuffer: ActorRef)(implicit materializer: Materializer) extends Topic[T] {

  // TODO: use ServiceInfo's name as a default value.
  def subscribe = new SubscriberStub("default", topicBuffer)

  class SubscriberStub(groupId: String, topicBuffer: ActorRef)(implicit materializer: Materializer)
    extends InternalSubscriberStub[T](groupId, topicBuffer)(materializer) with Subscriber[T] {

    override def withGroupId(groupId: String): Subscriber[T] = new SubscriberStub(groupId, topicBuffer)
    override def atMostOnceSource(): Source[T, _] = super.mostOnceSource.asJava
    override def atLeastOnce(flow: Flow[T, Done, _]): CompletionStage[Done] = toJava(super.leastOnce(flow.asScala))
  }
} 

//设置package包名称以及导入依赖的类
package com.lightbend.lagom.javadsl.persistence.cassandra

import java.io.File
import java.util.concurrent.CompletionStage

import akka.persistence.cassandra.testkit.CassandraLauncher
import com.lightbend.lagom.javadsl.persistence.TestEntity.Evt
import com.lightbend.lagom.javadsl.persistence.{ ReadSideProcessor, TestEntityReadSide }
import com.lightbend.lagom.javadsl.persistence.cassandra.testkit.TestUtil
import com.lightbend.lagom.javadsl.persistence.multinode.{ AbstractClusteredPersistentEntityConfig, AbstractClusteredPersistentEntitySpec }
import com.typesafe.config.{ Config, ConfigFactory }

object CassandraClusteredPersistentEntityConfig extends AbstractClusteredPersistentEntityConfig {
  override def additionalCommonConfig(databasePort: Int): Config =
    TestUtil.persistenceConfig("ClusteredPersistentEntitySpec", databasePort)
}

class CassandraClusteredPersistentEntitySpecMultiJvmNode1 extends CassandraClusteredPersistentEntitySpec
class CassandraClusteredPersistentEntitySpecMultiJvmNode2 extends CassandraClusteredPersistentEntitySpec
class CassandraClusteredPersistentEntitySpecMultiJvmNode3 extends CassandraClusteredPersistentEntitySpec

class CassandraClusteredPersistentEntitySpec extends AbstractClusteredPersistentEntitySpec(CassandraClusteredPersistentEntityConfig) {

  import CassandraClusteredPersistentEntityConfig._

  override protected def atStartup() {
    runOn(node1) {
      val cassandraDirectory = new File("target/" + system.name)
      CassandraLauncher.start(cassandraDirectory, "lagom-test-embedded-cassandra.yaml", clean = true, port = databasePort)
      TestUtil.awaitPersistenceInit(system)
    }
    enterBarrier("cassandra-started")

    super.atStartup()
  }

  override protected def afterTermination() {
    super.afterTermination()

    CassandraLauncher.stop()
  }

  def testEntityReadSide = injector.instanceOf[TestEntityReadSide]

  override protected def getAppendCount(id: String): CompletionStage[java.lang.Long] =
    testEntityReadSide.getAppendCount(id)

  override protected def readSideProcessor: Class[_ <: ReadSideProcessor[Evt]] = classOf[TestEntityReadSide.TestEntityReadSideProcessor]
} 

//设置package包名称以及导入依赖的类
package com.lightbend.lagom.javadsl.persistence.cassandra

import java.util.concurrent.CompletionStage

import com.google.inject.Guice
import com.lightbend.lagom.internal.javadsl.persistence.cassandra.{ CassandraPersistentEntityRegistry, CassandraReadSideImpl, JavadslCassandraOffsetStore }
import com.lightbend.lagom.internal.persistence.ReadSideConfig
import com.lightbend.lagom.internal.persistence.cassandra.CassandraReadSideSettings
import com.lightbend.lagom.javadsl.persistence._
import com.typesafe.config.ConfigFactory

import scala.concurrent.duration._

object CassandraReadSideSpec {

  val defaultConfig = ConfigFactory.parseString("akka.loglevel = INFO")
  val noAutoCreateConfig = ConfigFactory.parseString("lagom.persistence.read-side.cassandra.tables-autocreate = false")
}

class CassandraReadSideSpec extends CassandraPersistenceSpec(CassandraReadSideSpec.defaultConfig) with AbstractReadSideSpec {
  import system.dispatcher

  private lazy val injector = Guice.createInjector()
  override protected lazy val persistentEntityRegistry = new CassandraPersistentEntityRegistry(system, injector)

  private lazy val testSession: CassandraSession = new CassandraSession(system)
  private lazy val testCasReadSideSettings: CassandraReadSideSettings = new CassandraReadSideSettings(system)
  private lazy val offsetStore = new JavadslCassandraOffsetStore(system, testSession, testCasReadSideSettings, ReadSideConfig())
  private lazy val cassandraReadSide = new CassandraReadSideImpl(system, testSession, offsetStore, null, injector)

  override def processorFactory(): ReadSideProcessor[TestEntity.Evt] =
    new TestEntityReadSide.TestEntityReadSideProcessor(cassandraReadSide, testSession)

  private lazy val readSide = new TestEntityReadSide(testSession)

  override def getAppendCount(id: String): CompletionStage[java.lang.Long] = readSide.getAppendCount(id)

  override def afterAll(): Unit = {
    persistentEntityRegistry.gracefulShutdown(5.seconds)
    super.afterAll()
  }

}

class CassandraReadSideAutoCreateSpec extends CassandraPersistenceSpec(CassandraReadSideSpec.noAutoCreateConfig) {
  import system.dispatcher

  private lazy val testSession: CassandraSession = new CassandraSession(system)
  private lazy val testCasReadSideSettings: CassandraReadSideSettings = new CassandraReadSideSettings(system)
  private lazy val offsetStore = new JavadslCassandraOffsetStore(system, testSession, testCasReadSideSettings, ReadSideConfig())

  "A Cassandra Read-Side" must {
    "not send ClusterStartupTask message, so startupTask must return None" +
      "when 'lagom.persistence.read-side.cassandra.tables-autocreate' flag is 'false'" in {
        offsetStore.startupTask shouldBe None
      }
  }
} 

//设置package包名称以及导入依赖的类
package com.lightbend.lagom.internal.javadsl.persistence.jdbc

import java.util.concurrent.CompletionStage
import javax.inject.{ Inject, Singleton }

import com.lightbend.lagom.javadsl.persistence.jdbc.JdbcSession
import com.lightbend.lagom.javadsl.persistence.jdbc.JdbcSession.ConnectionFunction

import scala.compat.java8.FutureConverters._


@Singleton
final class JdbcSessionImpl @Inject() (slick: SlickProvider) extends JdbcSession {

  import slick.profile.api._

  override def withConnection[T](block: ConnectionFunction[T]): CompletionStage[T] = {
    slick.db.run {
      SimpleDBIO { ctx =>
        block(ctx.connection)
      }
    }.toJava
  }

  override def withTransaction[T](block: ConnectionFunction[T]): CompletionStage[T] = {
    slick.db.run {
      SimpleDBIO { ctx =>
        block(ctx.connection)
      }.transactionally
    }.toJava
  }

} 

//设置package包名称以及导入依赖的类
package com.lightbend.lagom.internal.javadsl.persistence.jpa

import java.util.concurrent.CompletionStage
import java.util.function.Supplier

import akka.actor.Scheduler
import akka.pattern.after

import scala.concurrent.duration.Duration.fromNanos
import scala.concurrent.duration.FiniteDuration
import scala.concurrent.{ ExecutionContext, Future }
import scala.util.control.NonFatal

// With thanks to https://gist.github.com/viktorklang/9414163
private[lagom] class Retry(delay: FiniteDuration, delayFactor: Double, maxRetries: Int) {
  def apply[T](op: => T)(implicit ec: ExecutionContext, s: Scheduler): Future[T] = {
    def iterate(nextDelay: FiniteDuration, remainingRetries: Int): Future[T] =
      Future(op) recoverWith {
        case NonFatal(throwable) if remainingRetries > 0 => {
          onRetry(throwable, nextDelay, remainingRetries)
          after(nextDelay, s)(iterate(finiteMultiply(nextDelay, delayFactor), remainingRetries - 1))
        }
      }

    iterate(delay, maxRetries)
  }

  // For convenient use from Java 8
  def retry[T](op: Supplier[T])(implicit ec: ExecutionContext, s: Scheduler): CompletionStage[T] = {
    import scala.compat.java8.FutureConverters._

    apply(op.get()).toJava
  }

  protected def onRetry(throwable: Throwable, delay: FiniteDuration, remainingRetries: Int): Unit = ()

  private def finiteMultiply(duration: FiniteDuration, factor: Double): FiniteDuration =
    fromNanos((duration.toNanos * factor).toLong)
}

private[lagom] object Retry {
  def apply[T](delay: FiniteDuration, delayFactor: Double, maxRetries: Int)(op: => T)(implicit ec: ExecutionContext, s: Scheduler): Future[T] =
    (new Retry(delay, delayFactor, maxRetries))(op)
} 

//设置package包名称以及导入依赖的类
package com.lightbend.lagom.javadsl.client

import java.net.URI
import java.util.concurrent.{ CompletionStage, TimeUnit }
import java.util.function.Supplier
import com.typesafe.config.ConfigFactory
import org.scalatest.{ Matchers, WordSpec }

import scala.compat.java8.OptionConverters._

class ConfigurationServiceLocatorSpec extends WordSpec with Matchers {

  val serviceLocator = new ConfigurationServiceLocator(ConfigFactory.parseString(
    """
      |lagom.services {
      |  foo = "http://localhost:10001"
      |  bar = "http://localhost:10002"
      |}
    """.stripMargin
  ), new CircuitBreakersPanel {
    override def withCircuitBreaker[T](id: String, body: Supplier[CompletionStage[T]]): CompletionStage[T] = body.get()
  })

  def locate(serviceName: String) =
    serviceLocator.locate(serviceName).toCompletableFuture.get(10, TimeUnit.SECONDS).asScala

  "ConfigurationServiceLocator" should {
    "return a found service" in {
      locate("foo") should contain(URI.create("http://localhost:10001"))
      locate("bar") should contain(URI.create("http://localhost:10002"))
    }
    "return none for not found service" in {
      locate("none") shouldBe None
    }
  }

} 

//设置package包名称以及导入依赖的类
package com.elkozmon.akka.firebase.javadsl

import java.util.concurrent.CompletionStage

import akka.Done
import akka.stream.javadsl.Source
import com.elkozmon.akka.firebase.internal.JavaWrapperConsumerControl
import com.elkozmon.akka.firebase.{Document, scaladsl}
import com.google.firebase.database.DatabaseReference

import scala.compat.java8.FutureConverters._


  def asyncSource(
    sourceNode: DatabaseReference,
    bufferSize: Int
  ): Source[CompletionStage[Document], Control] =
    scaladsl.Consumer
      .asyncSource(sourceNode, bufferSize)
      .map(_.toJava)
      .mapMaterializedValue(new JavaWrapperConsumerControl(_))
      .asJava
} 

//设置package包名称以及导入依赖的类
// Copyright (c) Microsoft. All rights reserved.

package com.microsoft.azure.iot.iothubreact.sinks

import java.util.concurrent.CompletionStage

import akka.Done
import akka.japi.function.Procedure
import akka.stream.javadsl.{Sink ? JavaSink}
import akka.stream.scaladsl.{Sink ? ScalaSink}
import com.microsoft.azure.iot.iothubreact.config.{ConnectConfiguration, IConnectConfiguration}
import com.microsoft.azure.iot.iothubreact.{Logger, MessageFromDevice, MessageToDevice}
import com.microsoft.azure.sdk.iot.service.{IotHubServiceClientProtocol, ServiceClient}

object MessageToDeviceSink {
  def apply(): MessageToDeviceSink = new MessageToDeviceSink()

  def apply(config: IConnectConfiguration): MessageToDeviceSink = new MessageToDeviceSink(config)
}


class MessageToDeviceSink(config: IConnectConfiguration)
  extends ISink[MessageToDevice]
    with Logger {

  // Parameterless ctor
  def this() = this(ConnectConfiguration())

  private[iothubreact] val protocol     = IotHubServiceClientProtocol.AMQPS
  private[iothubreact] val timeoutMsecs = 15000

  private[this] val connString    = s"HostName=${config.accessHostname};" +
    s"SharedAccessKeyName=${config.accessPolicy};" +
    s"SharedAccessKey=${config.accessKey}"
  private[this] val serviceClient = ServiceClient.createFromConnectionString(connString, protocol)

  log.info("Connecting client to ${} ...", config.accessHostname)
  serviceClient.open()

  def scalaSink(): ScalaSink[MessageToDevice, scala.concurrent.Future[Done]] =
    ScalaSink.foreach[MessageToDevice](
      m ? {
        log.info("Sending message to device {}", m.deviceId)
        serviceClient.sendAsync(m.deviceId, m.message)
      })

  def javaSink(): JavaSink[MessageToDevice, CompletionStage[Done]] =
    JavaSink.foreach[MessageToDevice] {
      JavaSinkProcedure
    }

  // Required for Scala 2.11
  private[this] object JavaSinkProcedure extends Procedure[MessageToDevice] {
    @scala.throws[Exception](classOf[Exception])
    override def apply(m: MessageToDevice): Unit = {
      log.info("Sending message to device " + m.deviceId)
      serviceClient.sendAsync(m.deviceId, m.message)
    }
  }
} 

//设置package包名称以及导入依赖的类
package one.aleph.akkzure.queue.javadsl

import com.microsoft.azure.storage.queue.{ CloudQueue, CloudQueueMessage }
import one.aleph.akkzure.queue.{ AzureQueueSinkFunctions, FlowMapECStage, DeleteOrUpdateMessage }
import akka.stream.javadsl.Sink
import akka.Done
import scala.concurrent.{ Future, ExecutionContext }
import java.util.concurrent.CompletionStage

object AzureQueueSink {
  def create(cloudQueue: CloudQueue, maxInFlight: Int): Sink[CloudQueueMessage, CompletionStage[Done]] = {
    fromFunction(AzureQueueSinkFunctions.addMessage(cloudQueue)(_)(_), maxInFlight)
  }

  def create(cloudQueue: CloudQueue): Sink[CloudQueueMessage, CompletionStage[Done]] = {
    create(cloudQueue, 4)
  }

  private[javadsl] def fromFunction[T](f: (T, ExecutionContext) => Future[Done], maxInFlight: Int): Sink[T, CompletionStage[Done]] = {
    import one.aleph.akkzure.queue.scaladsl.{ AzureQueueSink => AzureQueueSinkScalaDSL }
    import scala.compat.java8.FutureConverters._
    AzureQueueSinkScalaDSL.fromFunction(f, maxInFlight).mapMaterializedValue(_.toJava).asJava
  }
}

object AzureQueueDeleteSink {
  def create(cloudQueue: CloudQueue, maxInFlight: Int): Sink[CloudQueueMessage, CompletionStage[Done]] = {
    AzureQueueSink.fromFunction(AzureQueueSinkFunctions.deleteMessage(cloudQueue)(_)(_), maxInFlight)
  }

  def create(cloudQueue: CloudQueue): Sink[CloudQueueMessage, CompletionStage[Done]] = {
    create(cloudQueue, 4)
  }
}

object AzureQueueDeleteOrUpdateSink {
  def create(cloudQueue: CloudQueue, maxInFlight: Int): Sink[(CloudQueueMessage, DeleteOrUpdateMessage), CompletionStage[Done]] = {
    AzureQueueSink.fromFunction((input, ec) =>
      AzureQueueSinkFunctions.deleteOrUpdateMessage(cloudQueue)(input._1, input._2)(ec), maxInFlight)
  }
  def create(cloudQueue: CloudQueue): Sink[(CloudQueueMessage, DeleteOrUpdateMessage), CompletionStage[Done]] = {
    create(cloudQueue, 4)
  }
} 

//设置package包名称以及导入依赖的类
package com.lightbend.lagom.dns

import java.net.URI
import java.util.Optional
import java.util.concurrent.CompletionStage
import java.util.function.{ Function => JFunction }
import javax.inject.{ Inject, Named }

import akka.actor.{ ActorRef, ActorSystem }
import akka.pattern.ask
import com.lightbend.lagom.javadsl.api.{ Descriptor, ServiceLocator }
import com.lightbend.dns.locator.{ Settings, ServiceLocator => ServiceLocatorService }

import scala.compat.java8.FutureConverters._
import scala.compat.java8.OptionConverters._
import scala.concurrent.{ ExecutionContext, Future }


class DnsServiceLocator @Inject() (
    @Named("ServiceLocatorService") serviceLocatorService: ActorRef,
    system: ActorSystem,
    implicit val ec: ExecutionContext) extends ServiceLocator {

  val settings = Settings(system)

  private def locateAsScala(name: String): Future[Option[URI]] =
    serviceLocatorService
      .ask(ServiceLocatorService.GetAddress(name))(settings.resolveTimeout1 + settings.resolveTimeout1 + settings.resolveTimeout2)
      .mapTo[ServiceLocatorService.Addresses]
      .map {
        case ServiceLocatorService.Addresses(addresses) =>
          addresses
            .headOption
            .map(sa => new URI(sa.protocol, null, sa.host, sa.port, null, null, null))
      }

  override def locate(name: String, serviceCall: Descriptor.Call[_, _]): CompletionStage[Optional[URI]] =
    locateAsScala(name).map(_.asJava).toJava

  override def doWithService[T](name: String, serviceCall: Descriptor.Call[_, _], block: JFunction[URI, CompletionStage[T]]): CompletionStage[Optional[T]] =
    locateAsScala(name).flatMap(uriOpt => {
      uriOpt.fold(Future.successful(Optional.empty[T])) { uri =>
        block.apply(uri)
          .toScala
          .map(v => Optional.of(v))
      }
    }).toJava
}