This is an automated email from the ASF dual-hosted git repository.
chl501 pushed a commit to branch componentization
in repository https://gitbox.apache.org/repos/asf/hama.git
The following commit(s) were added to refs/heads/componentization by this push:
new aa8e9fc Add Subscription, Publication operations, and a test case.
aa8e9fc is described below
commit aa8e9fc504b1fbd11125da2e3e7aceab3812907b
Author: Chiahung Lin <[email protected]>
AuthorDate: Sun Apr 28 17:28:42 2019 +0200
Add Subscription, Publication operations, and a test case.
---
.../org/apache/hama/membership/Communicator.scala | 115 +++++++++++++++++++--
.../scala/org/apache/hama/membership/package.scala | 2 -
.../apache/hama/membership/CommunicatorSpec.scala | 59 +++++++++++
3 files changed, 164 insertions(+), 12 deletions(-)
diff --git
a/membership/src/main/scala/org/apache/hama/membership/Communicator.scala
b/membership/src/main/scala/org/apache/hama/membership/Communicator.scala
index 26f0117..5705baa 100644
--- a/membership/src/main/scala/org/apache/hama/membership/Communicator.scala
+++ b/membership/src/main/scala/org/apache/hama/membership/Communicator.scala
@@ -31,6 +31,7 @@ import org.agrona.BufferUtil
import org.agrona.DirectBuffer
import org.agrona.concurrent.BusySpinIdleStrategy
import org.agrona.concurrent.SigInt
+import org.agrona.concurrent.ShutdownSignalBarrier
import org.agrona.concurrent.UnsafeBuffer
import org.apache.hama.logging.Logging
import org.apache.hama.util.Utils._
@@ -44,13 +45,115 @@ import scalaz.zio._
trait Communicator
object Communicator {
- def driver: Task[MediaDriver] = ZIO.effect {
+ val localhost = "localhost"
+
+ final case object BackPressured extends RuntimeException (
+ "Failure due to back pressure!"
+ )
+
+ final case object NotConnected extends RuntimeException (
+ "Failure because publisher is not connected to subscriber!"
+ )
+
+ final case object AdminAction extends RuntimeException (
+ "Failure because of administration action!"
+ )
+
+ final case object Closed extends RuntimeException (
+ "Failure because publication is closed!"
+ )
+
+ final case object MaxPositionExceeded extends RuntimeException (
+ "Failure due to publication reaching max position!"
+ )
+
+ final case object UnknownReason extends RuntimeException (
+ "Failure due to unknown reason!"
+ )
+
+ case class Channel(host: String = localhost, port: Int = 12345) {
+ override def toString(): String = s"aeron:udp?endpoint=$host:$port"
+ }
+
+ implicit class SubscriberOps(aeron: Task[Aeron]) {
+ def subscriber(channel: Channel = Channel(), streamId: Int = 10):
+ Task[Subscription] = aeron.flatMap { a =>
+ Task.effect(a.addSubscription(channel.toString, streamId))
+ }
+ }
+
+ implicit class ReceiveOps(subscriber: Task[Subscription]) {
+ def receive(f: (DirectBuffer, Int, Int, Header) => Boolean) = ZIO.effect {
+ var continuous = true
+ val handler = new FragmentAssembler({
+ (buffer: DirectBuffer, offset: Int, length: Int, header: Header) =>
+ val shouldContinuous = f(buffer, offset, length, header)
+ continuous = shouldContinuous
+ })
+ val idleStrategy = new BusySpinIdleStrategy()
+ while(continuous) subscriber.flatMap { s =>
+ val read = s.poll(handler, 10)
+ idleStrategy.idle(read)
+ ZIO.unit
+ }
+ this
+ }
+
+ def close() = subscriber.flatMap { sub => sub.close; ZIO.unit }
+
+ }
+
+ implicit class PublisherOps(aeron: Task[Aeron]) {
+ def publisher(channel: Channel = Channel(), streamId: Int = 10):
+ Task[Publication] = aeron.flatMap { a =>
+ Task.effect(a.addPublication(channel.toString, streamId))
+ }
+ }
+
+ implicit class PublishOps(publisher: Task[Publication]) {
+ def publish (
+ messageBytes: Array[Byte],
+ bufferCapacity: Int = 512,
+ boundaryAlighment: Int = CACHE_LINE_LENGTH,
+ deadline: Long = (System.nanoTime + 3.seconds.toNanos),
+ sleep: () => Unit = { () => Thread.sleep(1) }
+ ): ZIO[Any, Throwable, PublishOps] = publisher.flatMap { pub =>
+ while(!pub.isConnected) {
+ if(System.nanoTime >= deadline) return ZIO.fail(new RuntimeException (
+ s"Publication can't connect: ${pub.toString}"
+ ))
+ sleep()
+ }
+ val buffer = new UnsafeBuffer(BufferUtil.allocateDirectAligned (
+ bufferCapacity, boundaryAlighment
+ ))
+ buffer.putBytes(0, messageBytes)
+ val result = pub.offer(buffer, 0, messageBytes.length)
+ result match {
+ case res if res < 0L && BACK_PRESSURED == res =>
ZIO.fail(BackPressured)
+ case res if res < 0L && NOT_CONNECTED == res => ZIO.fail(NotConnected)
+ case res if res < 0L && ADMIN_ACTION == res => ZIO.fail(AdminAction)
+ case res if res < 0L && CLOSED == res => ZIO.fail(Closed)
+ case res if res < 0L && MAX_POSITION_EXCEEDED == res => ZIO.fail(
+ MaxPositionExceeded
+ )
+ case res if res < 0L => ZIO.fail(UnknownReason)
+ case _ => ZIO.succeed(this)
+ }
+ }
+
+ def close() = publisher.flatMap { pub => pub.close; ZIO.unit }
+ }
+
+ def driver(barrier: ShutdownSignalBarrier): Task[MediaDriver] = ZIO.effect {
val context = new MediaDriver.Context().
threadingMode(DEDICATED).
conductorIdleStrategy(new BusySpinIdleStrategy).
receiverIdleStrategy(new BusySpinIdleStrategy).
senderIdleStrategy(new BusySpinIdleStrategy)
- MediaDriver.launch(context)
+ val driver = MediaDriver.launch(context)
+ barrier.await
+ driver
}
def aeron: Task[Aeron] = ZIO.effect {
@@ -58,12 +161,4 @@ object Communicator {
Aeron.connect(context)
}
- def subscriber(aeron: Task[Aeron]): Task[Subscription] = aeron.flatMap { a =>
- Task.effect(a.addSubscription("aeron:udp?endpoint=localhost:12345", 10))
- }
-
- def publisher(aeron: Task[Aeron]): Task[Publication] = aeron.flatMap { a =>
- Task.effect(a.addPublication("aeron:udp?endpoint=localhost:12344", 10))
- }
-
}
diff --git a/membership/src/main/scala/org/apache/hama/membership/package.scala
b/membership/src/main/scala/org/apache/hama/membership/package.scala
index 2c7c29f..e01e7a3 100644
--- a/membership/src/main/scala/org/apache/hama/membership/package.scala
+++ b/membership/src/main/scala/org/apache/hama/membership/package.scala
@@ -23,6 +23,4 @@ package object membership {
type ValidatedResult[A] = ValidatedNel[Throwable, A]
- val localhost = "localhost"
-
}
diff --git
a/membership/src/test/scala/org/apache/hama/membership/CommunicatorSpec.scala
b/membership/src/test/scala/org/apache/hama/membership/CommunicatorSpec.scala
new file mode 100644
index 0000000..49740b8
--- /dev/null
+++
b/membership/src/test/scala/org/apache/hama/membership/CommunicatorSpec.scala
@@ -0,0 +1,59 @@
+/**
+ * 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.hama.membership
+
+import io.aeron.logbuffer.Header
+import org.scalatest._
+import org.agrona.DirectBuffer
+import org.agrona.concurrent.ShutdownSignalBarrier
+import scalaz.zio._
+
+class CommunicatorSpec extends FlatSpec with Matchers {
+
+ import Communicator._
+
+ "Communicator" should "" in {
+
+
+ val barrier = new ShutdownSignalBarrier
+ val dio = driver(barrier)
+ val a = aeron
+ val subio = a.subscriber().receive {
+ (buffer: DirectBuffer, offset: Int, length: Int, header: Header) =>
+ val data = new Array[Byte](length)
+ buffer.getBytes(offset, data)
+ val hello = new String(data)
+ false
+ }.flatMap(_.close)
+ val pubio = a.publisher().publish(messageBytes =
"hello".getBytes).flatMap(_.close)
+
+ val handler = for {
+ forked_d <- dio.fork
+ forked_sub <- subio.fork
+ forked_pub <- pubio.fork
+ forked_b <- ZIO.effect { barrier.signal; ZIO.unit }.fork
+ _ <- forked_b.join
+ p <- forked_pub.join
+ s <- forked_sub.join
+ d <- forked_d.join
+ } yield (p, s, d)
+
+ val runtime = new DefaultRuntime {}
+ runtime.unsafeRun { handler.run }
+ }
+}
