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 c346069 Group Subscription, and Publication functuions.
c346069 is described below
commit c3460697e29b52a84b42729c556a3237d1c8da2b
Author: Chiahung Lin <[email protected]>
AuthorDate: Wed May 1 23:37:41 2019 +0200
Group Subscription, and Publication functuions.
---
.../org/apache/hama/membership/Communicator.scala | 195 +++++++++++----------
.../apache/hama/membership/CommunicatorSpec.scala | 10 +-
2 files changed, 106 insertions(+), 99 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 ba6a3db..1c51ae3 100644
--- a/membership/src/main/scala/org/apache/hama/membership/Communicator.scala
+++ b/membership/src/main/scala/org/apache/hama/membership/Communicator.scala
@@ -73,6 +73,101 @@ object Communicator {
"Failure due to unknown reason!"
)
+ case class Subscriber(subscription: Task[Subscription]) {
+
+ def receive (
+ f: (DirectBuffer, Int, Int, Header) => Boolean
+ ) = subscription.flatMap { s =>
+ 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) {
+ val read = s.poll(handler, 10)
+ idleStrategy.idle(read)
+ }
+ subscription
+ }
+
+ }
+
+ case class Publisher(publication: Task[Publication]) {
+
+ def publish (
+ messageBytes: Array[Byte],
+ bufferCapacity: Int = 512,
+ boundaryAlighment: Int = CACHE_LINE_LENGTH,
+ countLimit: Int = 3,
+ deadline: Long = (System.nanoTime + 3.seconds.toNanos),
+ sleep: () => Unit = { () => Thread.sleep(1) }
+ ) = {
+
+ def isConnected (publisher: Publication)(
+ implicit _deadline: Long = (System.nanoTime + 3.seconds.toNanos),
+ _countLimit: Int = 3,
+ _sleep: () => Unit = { () => Thread.sleep(1) }
+ ): Boolean = {
+ @tailrec
+ def _isConnected(pub: Publication, times: Int = 0): Boolean = {
+ val connected = pub.isConnected
+ if(_countLimit > times && (false == connected)) {
+ if(System.nanoTime >= _deadline) false else {
+ _sleep()
+ _isConnected(pub, times + 1)
+ }
+ } else connected
+ }
+ _isConnected(publisher)
+ }
+
+ def _publish (
+ messageBytes: Array[Byte],
+ bufferCapacity: Int = 512,
+ boundaryAlighment: Int = CACHE_LINE_LENGTH,
+ countLimit: Int = 3,
+ deadline: Long = (System.nanoTime + 3.seconds.toNanos),
+ sleep: () => Unit = { () => Thread.sleep(1) }
+ ) = publication.flatMap { p =>
+ implicit val _deadline = deadline
+ implicit val _countLimit = countLimit
+ implicit val _sleep = sleep
+ isConnected(p) match {
+ case true =>
+ val buffer = new UnsafeBuffer(BufferUtil.allocateDirectAligned (
+ bufferCapacity, boundaryAlighment
+ ))
+ buffer.putBytes(0, messageBytes)
+ val result = p.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)
+ }
+ case false => ZIO.fail(new RuntimeException("Not Connected to
Subscriber!"))
+ }
+ }
+
+ _publish (
+ messageBytes = messageBytes,
+ bufferCapacity = bufferCapacity,
+ boundaryAlighment = boundaryAlighment,
+ countLimit = countLimit,
+ deadline = deadline,
+ sleep = sleep
+ )
+ }
+
+ }
+
val localhost = "localhost"
def driver(barrier: ShutdownSignalBarrier): Task[MediaDriver] = ZIO.effect {
@@ -91,100 +186,14 @@ object Communicator {
Aeron.connect(context)
}
- def subscriber(aeron: Task[Aeron]) = aeron.flatMap { a =>
- ZIO.effect { a.addSubscription(Channel().toString, 10) }
- }
-
- def receive (
- sub: Task[Subscription],
- f: (DirectBuffer, Int, Int, Header) => Boolean
- ) = sub.flatMap { s =>
- var continuous = true
- val handler = new FragmentAssembler({
- (buffer: DirectBuffer, offset: Int, length: Int, header: Header) =>
- val shouldContinuous = f(buffer, offset, length, header)
- continuous = shouldContinuous
+ def subscriber(channel: Channel = Channel(), streamId: Int = 10) =
+ Subscriber(aeron.flatMap { _aeron =>
+ ZIO.effect(_aeron.addSubscription(channel.toString, streamId))
})
- val idleStrategy = new BusySpinIdleStrategy()
- while(continuous) {
- val read = s.poll(handler, 10)
- idleStrategy.idle(read)
- }
- sub
- }
-
- def publisher(aeron: Task[Aeron]) = aeron.flatMap { a =>
- ZIO.effect { a.addPublication(Channel().toString, 10) }
- }
- def publish(
- publisher: Task[Publication],
- messageBytes: Array[Byte],
- bufferCapacity: Int = 512,
- boundaryAlighment: Int = CACHE_LINE_LENGTH,
- countLimit: Int = 3,
- deadline: Long = (System.nanoTime + 3.seconds.toNanos),
- sleep: () => Unit = { () => Thread.sleep(1) }
- ) = {
-
- def isConnected (publication: Publication)(
- implicit _deadline: Long = (System.nanoTime + 3.seconds.toNanos),
- _countLimit: Int = 3,
- _sleep: () => Unit = { () => Thread.sleep(1) }
- ): Boolean = {
- @tailrec
- def _isConnected(pub: Publication, times: Int = 0): Boolean = {
- val connected = pub.isConnected
- if(_countLimit > times && (false == connected)) {
- if(System.nanoTime >= _deadline) false else {
- _sleep()
- _isConnected(pub, times + 1)
- }
- } else connected
- }
- _isConnected(publication)
- }
-
- def _publish (
- messageBytes: Array[Byte],
- bufferCapacity: Int = 512,
- boundaryAlighment: Int = CACHE_LINE_LENGTH,
- countLimit: Int = 3,
- deadline: Long = (System.nanoTime + 3.seconds.toNanos),
- sleep: () => Unit = { () => Thread.sleep(1) }
- ) = publisher.flatMap { p =>
- implicit val _deadline = deadline
- implicit val _countLimit = countLimit
- implicit val _sleep = sleep
- isConnected(p) match {
- case true =>
- val buffer = new UnsafeBuffer(BufferUtil.allocateDirectAligned (
- bufferCapacity, boundaryAlighment
- ))
- buffer.putBytes(0, messageBytes)
- val result = p.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)
- }
- case false => ZIO.fail(new RuntimeException("Not Connected to
Subscriber!"))
- }
- }
- _publish (
- messageBytes = messageBytes,
- bufferCapacity = bufferCapacity,
- boundaryAlighment = boundaryAlighment,
- countLimit = countLimit,
- deadline = deadline,
- sleep = sleep
- )
- }
+ def publisher(channel: Channel = Channel(), streamId: Int = 10) =
+ Publisher(aeron.flatMap { _aeron =>
+ ZIO.effect(_aeron.addPublication(channel.toString, streamId))
+ })
}
diff --git
a/membership/src/test/scala/org/apache/hama/membership/CommunicatorSpec.scala
b/membership/src/test/scala/org/apache/hama/membership/CommunicatorSpec.scala
index f4fe826..5fefdcb 100644
---
a/membership/src/test/scala/org/apache/hama/membership/CommunicatorSpec.scala
+++
b/membership/src/test/scala/org/apache/hama/membership/CommunicatorSpec.scala
@@ -35,18 +35,15 @@ class CommunicatorSpec extends FlatSpec with Matchers {
var hello = ""
val barrier = new ShutdownSignalBarrier
val dio = driver(barrier).delay(ZDuration.fromScala(0 seconds))
- val sub = subscriber(aeron)
- val subio = receive(sub, {
+ val subio = subscriber().receive {
(buffer: DirectBuffer, offset: Int, length: Int, header: Header) =>
val data = new Array[Byte](length)
buffer.getBytes(offset, data)
hello = new String(data)
println("Received? "+hello)
false
- })
- val pub = publisher(aeron)
- val pubio = publish (
- publisher = pub,
+ }
+ val pubio = publisher().publish (
messageBytes = "hello".getBytes,
countLimit = 1024,
deadline = (System.nanoTime + 10.seconds.toNanos),
@@ -66,5 +63,6 @@ class CommunicatorSpec extends FlatSpec with Matchers {
val runtime = new DefaultRuntime {}
runtime.unsafeRun { handler.run }
+ assert("hello".equals(hello))
}
}
