Parapet

Process is a key abstraction in Parapet, any application must have a least one process. If you try to run an application w/o processes you will get an error saying that at least one process required. This section covers some useful features that we haven't seen yet, below you will find a shortlist of features:

• Predefined processes and reserved references
• Switching process behavior
• Direct process call
• Process combinators: and and or
• Testing your processes
• Basic patterns and tips

Creating the Process - #back to top

There are there several ways to create the Process.

Extending Process trait

  class GenericProcess[F[_]] extends Process[F] {

    import dsl._

    override def handle: Receive = {
      case Start => unit // TODO
    }
  }

or using a specific effect type, e.g. IO:

  class IOProcess extends Process[IO] {

    import dsl._

    override def handle: Receive = {
      case Start => unit // TODO
    }
  }

Using Process Builder

You can use Process Builder to create a stateless process:

Process.Builder setters:

• name sets a process name; undefined is a default value
• ref sets a process reference; jdk UUID is a default value
• bufferSize sets a process queue size; -1 - unbounded (by default) unless it's set globally usingParConfig#processBufferSize; valid values: -1; [1,Int.MaxValue)

Example:

  val process: Process[IO] = Process.builder[IO](ref => {
    case Start => dsl.unit
  }).ref(ProcessRef("process")) // sets a process reference
    .name("process") // sets a process name
    .bufferSize(1000) // set a process queue size limit
    .build

If you don't need to override any values then you can use Process#apply. Example:

  val process: Process[IO] = Process[IO](_ => {
    case Start => dsl.unit
  })

Predefined processes and reserved references - #back to top

Parapet has some reserved process references, e.g.: KernelRef(parapet-kernel), SystemRef(parapet-system), DeadLetterRef(parapet-deadletter), UndefinedRef(parapet-undefined). The general rule is that any reference that starts with parapet- prefix can be used by the platform code for any purpose. Parapet has a SystemProcess that cannot be overridden by users. SystemProcess is a starting point, i.e. it's created before any other process. Lifecycle event Start is sent by SystemProcess. Any event sent to the SystemProcess will be ignored and dropped. Don't try to send any events to SystemProcess b/c it can lead to unpredictable errors. DeadLetterProcess is another process that is created by default, although it can be overridden, for more details check DeadLetterProcess section under Event Handling.

Switching process behavior - #back to top

Sometimes it might be useful to dynamically switch a process behavior, e.g.: from uninitialized to ready state. Thankfully Process provides switch method that does exactly that.

Example (lazy server):

  // for some effect `F[_]`
  val server = new Process[F] {

    val init = eval(println("acquire resources: create socket and etc."))

    def ready: Receive = {
      case Request(data) => withSender(Success(data) ~> _)
      case Stop => eval(println("release resources: close socket and etc."))
    }

    def uninitialized: Receive = {
      case Start => unit // ignore Start event, wait for Init
      case Stop => unit // process is not initialized, do nothing
      case Init => init ++ switch(ready)
      case _ => withSender(Failure("process is not initialized", ErrorCodes.ProcessUninitialized) ~> _)
    }

    override def handle: Receive = uninitialized
  }

  // API

  object Init extends Event

  case class Request(data: Any) extends Event

  sealed trait Response extends Event
  case class Success(data: Any) extends Event
  case class Failure(data: Any, errorCode: Int) extends Event

  object ErrorCodes {
    val ProcessUninitialized  = 0
  }

A client which sends Request event w/o sending Init:

  val impatientClient = Process[F](_ => {
    case Start => Request("PING") ~> server
    case Success(_) => eval(println("that is not going to happen"))
    case f:Failure => eval(println(f))
  })

The code above will print:

Failure(process is not initialized,0)

A client which sends Init first and then Request:

  val humbleClient = Process[F](_ => {
    case Start => Seq(Init, Request("PING")) ~> server
    case Success(data) => eval(println(s"client receive response from server: $data"))
    case _:Failure => eval(println("that is not going to happen"))
  })

The code above will print:

acquire resources: create socket and etc.
client receive response from server: PING
release resources: close socket and etc.

switch is NOT an atomic operation, avoid using switch in concurrent flows because it may result in an error or lead to unpredictable behavior.

Not recommended:

  val process = new Process[F] {
    def ready: Receive = _

    override def handle: Receive = {
      case Init => fork(switch(ready)) // bad, may lead to unpredictable behaviour
    }
  }

If you need to switch behavior from a concurrent flow just send an event e.g. Swith(State.Ready) to itself. Process will eventually switch its behavior.

Recommended:

  val process = new Process[F] {
    def ready: Receive = _

    override def handle: Receive = {
      case Init => fork {
        eval(println("do some work in parallel"))
        Switch(Ready) ~> ref // notify the process that it's time to switch it's behaviour
      }
      case Switch(Ready) =>  switch(ready)
    }
  }

  sealed trait State
  object Ready extends State

  case class Switch(next: State) extends Event

Direct process call - #back to top

Sometimes it may be useful to call a process directly. Especially it's a common case for short living processes. For instance, you may want to create a process, call it and then abandon, garbage collector will do its job. However, if you try to send an event to a process that doesn't exist in the system you will receive Failure event with UnknownProcessException. This is where direct call comes to rescue.

Example:

  // API
  case class Sum(a: Int, b: Int) extends Event
  case class Result(value: Int) extends Event

  class Calculator[F[_]] extends Process[F] {
    override def handle: Receive = {
      case Sum(a, b) => withSender(Result(a + b - 1) ~> _) // yes, very poor calculator
    }
  }

  val student = Process[F](ref => {
    case Start => new Calculator().apply(ref, Sum(2, 2))
    case Result(value) => eval(println(s"2 + 2 = $value"))
  })

Output: 2 + 2 = 3

Note that apply method doesn't return a normal value rather it returns a program which will be executed as normal flow. In other words the following expressions are equivalent:

Sum(2, 2) ~> calculator <-> new Calculator().apply(ref, Sum(2, 2)) // where ref belongs to the same process in both cases

Also, you may be wondering why do we need to pass the process ref as an argument in the apply method. The reason is that the library needs to know the address of a sender so it can send a reply to it.

Process combinators - #back to top

Processes can be combined using two logical operators: or and and.

and - combines two processes by producing a new process with ref of the first process; combines flows iff handle function is defined for the given event in both processes. Sends an error to the sender if either of two processes isn't defined for the given event.

Example:

import cats.effect.IO
import io.parapet.CatsApp
import io.parapet.core.Event.Start
import io.parapet.core.{Event, Process}

object Example extends CatsApp {

  import dsl._

  case class Print(data: Any) extends Event

  override def processes: IO[Seq[Process[IO]]] =
    for {
      printerA <- IO.pure(Process[IO](_ => {
        case Print(data) => eval(println(s"printerA: $data"))
      }))
      printerB <- IO.pure(Process[IO](_ => {
        case Print(data) => eval(println(s"printerB: $data"))
      }))

      client <- IO.pure(Process[IO](ref => {
        case Start => printerA.and(printerB).apply(ref, Print("test"))
      }))

    } yield Seq(printerA, printerB, client)

}

printerA: test
printerB: test

If you want to register a combined process then you don't need to register printerA.

Example:

import cats.effect.IO
import io.parapet.CatsApp
import io.parapet.core.Event.Start
import io.parapet.core.{Event, Process}

object Example extends CatsApp {

  import dsl._

  case class Print(data: Any) extends Event

  override def processes: IO[Seq[Process[IO]]] =
    for {
      printerA <- IO.pure(Process[IO](_ => {
        case Print(data) => eval(println(s"printerA: $data"))
      }))
      printerB <- IO.pure(Process[IO](_ => {
        case Print(data) => eval(println(s"printerB: $data"))
      }))

      combined <- IO.pure(printerA.and(printerB))

      client <- IO.pure(Process[IO](_ => {
        case Start => Print("test") ~> combined
      }))

    } yield Seq(combined, printerB, client)

}

or - creates a new process with ref of the first process. A combined process refers to the first process if its handle is defined for the given event, otherwise, to the second process. Sends an error to the sender if neither process is defined for the given event.

Example:

import cats.effect.IO
import io.parapet.CatsApp
import io.parapet.core.Event.Start
import io.parapet.core.{Event, Process}

object Example extends CatsApp {

  import dsl._

  case class Print(data: Any) extends Event

  override def processes: IO[Seq[Process[IO]]] =
    for {
      printerA <- IO.pure(Process[IO](_ => {
        case Print(data: Int) => eval(println(s"printerA: $data"))
      }))
      printerB <- IO.pure(Process[IO](_ => {
        case Print(data: String) => eval(println(s"printerB: $data"))
      }))

      combined <- IO.pure(printerA.or(printerB))

      client <- IO.pure(Process[IO](_ => {
        case Start => Print("test") ~> combined ++ Print(1) ~> combined
      }))

    } yield Seq(combined, printerB, client)
}

printerB: test
printerA: 1

Testing your processes - #back to top

Integration tests in parapet written in a generic style that we discussed before so that the same tests can be run against any effect system. Let's try to write a simple test for a proxy process. The first thing you need to do is to add test-utils library into your project:

libraryDependencies += "io.parapet" %% "test-utils" % version

A simple proxy process that receives requests and forwards them to a service

  class Proxy(service: ProcessRef) extends Process[F] {
    override def handle: Receive = {
      case Request(data) => Request(s"proxy-$data") ~> service
    }
  }

Test for our Proxy:

import io.parapet.core.{Event, Process, ProcessRef}
import io.parapet.tests.intg.ProxySpec._
import io.parapet.testutils.{EventStore, IntegrationSpec}
import org.scalatest.FunSuite
import org.scalatest.Matchers._
import org.scalatest.OptionValues._


abstract class ProxySpec[F[_]] extends FunSuite with IntegrationSpec[F] {

  import dsl._

  test("proxy") {
    val eventStore = new EventStore[F, Event]
    val testService = Process(ref => {
      case req: Request => eval(eventStore.add(ref, req))
    })

    val proxy = new Proxy[F](testService.ref)

    val init = onStart(Request("req") ~> proxy)

    unsafeRun(eventStore.await(1, createApp(ct.pure(Seq(init, testService, proxy))).run))

    eventStore.get(testService.ref).headOption.value shouldBe Request("proxy-req")
  }

}

In order to run this test against Cats Effect IO you need to extend BasicCatsIOSpec:

import cats.effect.IO
import io.parapet.testutils.BasicCatsIOSpec

class ProxySpec extends io.parapet.tests.intg.ProxySpec[IO] with BasicCatsIOSpec

Basic patterns and tips - #back to top

Channel is a process that implements strictly synchronous request-reply dialog. The channel sends an event to a receiver and then waits for a response in one step, i.e. it blocks asynchronously until it receives a response. Doing any other sequence, e.g., sending two request or reply events in a row will return a failure to the sender.

Example for some F[_]:

  val server = new Process[F] {
    override def handle: Receive = {
      case Request(data) => withSender(sender => Response(s"echo: $data") ~> sender)
    }
  }

  val client = new Process[F] {

    lazy val ch = Channel[F]

    override def handle: Receive = {
      case Start => register(ref, ch) ++
        ch.send(Request("PING"), server.ref, {
          case scala.util.Success(Response(data)) => eval(println(data))
          case scala.util.Failure(err) => eval(println(s"server failed to process request. err: ${err.getMessage}"))
        })

    }
  }


  case class Request(data: Any) extends Event

  case class Response(data: Any) extends Event

echo: PING

There are some scenarios when a process may receive a Failure event:

When a target process failed to handle an event sent by another process.

Example:

  // for some effect F[_]
  val faultyServer = Process.builder[F](_ => {
    case Request(_) => eval(throw new RuntimeException("server is down"))
  }).ref(ProcessRef("server")).build

  val client = Process.builder[F](_ => {
    case Start => Request("PING") ~> faultyServer
    case Failure(Envelope(me, event, receiver), EventHandlingException(errMsg, cause)) => eval {
      println(s"self: $me")
      println(s"event: $event")
      println(s"receiver: $receiver")
      println(s"errMsg: $errMsg")
      println(s"cause: ${cause.getMessage}")
    }
  }).ref(ProcessRef("client")).build

The code above will output:

self: client
event: Request(PING)
receiver: server
errMsg: process [name=undefined, ref=server] has failed to handle event: Request(PING)
cause: server is down

EventHandlingException indicates that a receiver process failed to handle an event.

When a process event queue is full.

It's possible when a process experiencing performance degradation due to heavy load.

Example:

For this example we need to tweak SchedulerConfig:

queueSize = 10000
processQueueSize = 100

  // for some effect F[_]
  val slowServer = Process.builder[F](_ => {
    case Request(_) => eval(while (true) {}) // very slow process... NEVER WRITE SUCH CODE
  }).ref(ProcessRef("server")).build

  val client = Process.builder[F](_ => {
    case Start =>
      generateRequests(1000) ~> slowServer
    case Failure(Envelope(me, event, receiver), EventDeliveryException(errMsg, cause)) => eval {
      println(s"self: $me")
      println(s"event: $event")
      println(s"receiver: $receiver")
      println(s"errMsg: $errMsg")
      println(s"cause: ${cause.getMessage}")
      println("=====================================================")
    }
  }).ref(ProcessRef("client")).build

  def generateRequests(n: Int): Seq[Event] = {
    (0 until n).map(Request)
  }

The code above will print a dozens of lines, four lines per Failure event:

client sent events
self: client
event: Request(101)
receiver: server
errMsg: System failed to deliver an event to process [name=undefined, ref=server]
cause: process [name=undefined, ref=server] event queue is full
=====================================================
self: client
event: Request(102)
receiver: server
errMsg: System failed to deliver an event to process [name=undefined, ref=server]
cause: process [name=undefined, ref=server] event queue is full
=====================================================
self: client
event: Request(103)
receiver: server
errMsg: System failed to deliver an event to process [name=undefined, ref=server]
cause: process [name=undefined, ref=server] event queue is full
=====================================================

EventDeliveryException indicates that the system failed to deliver an event. Handling such types of errors may be useful for runtime analysis, e.g. a sender process might consider lowering event send rate or even stop sending events to let a target process to finish processing pending events. It's worth noting that you should avoid any long-running computations when processing Failure events because it could lead to cascading failures.

A process event handler isn't defined for some events.

Example:

  // for some effect F[_]
  val uselessService = Process.builder[F](_ => {
    case Start => unit
    case Stop => unit
  }).ref(ProcessRef("server")).build

  val client = Process.builder[F](_ => {
    case Start =>
      Request("PING") ~> uselessService
    case Failure(Envelope(me, event, receiver), EventMatchException(errMsg)) => eval {
      println(s"self: $me")
      println(s"event: $event")
      println(s"receiver: $receiver")
      println(s"errMsg: $errMsg")
    }
  }).ref(ProcessRef("client")).build

The code above will print:

self: client
event: Request(PING)
receiver: server
errMsg: process [name=undefined, ref=server] handler is not defined for event: Request(PING)

A process doesn't exist in Parapet system.

  // for some effect F[_]
  val unknownService = Process.builder[F](_ => {
    case Start => unit
    case Stop => unit
  }).ref(ProcessRef("server")).build

  val client = Process.builder[F](_ => {
    case Start =>
      Request("PING") ~> unknownService
    case Failure(Envelope(me, event, receiver), UnknownProcessException(errMsg)) => eval {
      println(s"self: $me")
      println(s"event: $event")
      println(s"receiver: $receiver")
      println(s"errMsg: $errMsg")
    }
  }).ref(ProcessRef("client")).build

The code above will print:

self: client
event: Request(PING)
receiver: server
errMsg: there is no such process with id=server registered in the system

Final notes regarding error handling:

• All Failure events sent by parapet-system process (if you are curious you can check it by yourself using withSender).
• If a process has no error handling then Failure event will be sent to DeadLetterProcess. More about DeadLetterProcess you will find below

DeadLetterProcess #back to top

The library by default provides an implementation of DeadLetterProcess which just logs failures. Although it might be not very practical, for instance, you may prefer to store failures into a database for further analyses. The library allows providing a custom implementation of DeadLetterProcess.

Example using CatsApp:

import cats.effect.IO
import io.parapet.CatsApp
import io.parapet.core.Event.{DeadLetter, Start}
import io.parapet.core.processes.DeadLetterProcess
import io.parapet.core.{Event, Process, ProcessRef}


object CustomDeadLetterProcessDemo extends CatsApp {

  import dsl._

  override def deadLetter: IO[DeadLetterProcess[IO]] = IO.pure {
    new DeadLetterProcess[IO] {
      override def handle: Receive = {
        // can be stored in database
        case DeadLetter(envelope, error) => eval {
          println(s"sender: ${envelope.sender}")
          println(s"receiver: ${envelope.receiver}")
          println(s"event: ${envelope.event}")
          println(s"errorType: ${error.getClass.getSimpleName}")
          println(s"errorMsg: ${error.getMessage}")
        }
      }
    }
  }

  val faultyServer = Process.builder[IO](_ => {
    case Request(_) => eval(throw new RuntimeException("server is down"))
  }).ref(ProcessRef("server")).build

  val client = Process.builder[IO](_ => {
    case Start => Request("PING") ~> faultyServer
    // no error handling
  }).ref(ProcessRef("client")).build

  override def processes: IO[Seq[Process[IO]]] = IO {
    Seq(client, faultyServer)
  }

  case class Request(data: Any) extends Event

}

The code above will print:

sender: client
receiver: server
event: Request(PING)
errorType: EventHandlingException
errorMsg: process [name=undefined, ref=server] has failed to handle event: Request(PING)

EventLog can be used to store events on disk. Latter, events can be retrieved and resubmitted. In a case, the event queue is full all unsubmitted events will be redirected to EventLog.

The default implementation just logs such events. In future releases, more practical implementation will be provided.

Parapet system can be configured by providing an instance of ParConfig.

Example:

import cats.effect.IO
import io.parapet.core.Parapet.ParConfig
import io.parapet.{CatsApp, core}

object ConfigExample extends CatsApp{
  override def processes: IO[Seq[core.Process[IO]]] = _

  override val config: ParConfig = ParConfig(...)
}

ParConfig has the following properties

• processBufferSize sets a buffer size limit for all processes. If a process created with a different value then that value will be used.

  Examples:

  Example 1:

    override val config: ParConfig = ParConfig(-1, SchedulerConfig.default)
  
    val processA = Process[IO](_ => {
      case _ => unit
    })
  
    val processB = Process.builder[IO](_ => {
      case _ => unit
    }).bufferSize(1000).build
                                          

  processA will be created with unbounded buffer

  processB will be created with buffer size limit equals 1000

  Example 2:

    override val config: ParConfig = ParConfig(5000, SchedulerConfig.default)
  
    val processA = Process[IO](_ => {
      case _ => unit
    })
  
    val processB = Process.builder[IO](_ => {
      case _ => unit
    }).bufferSize(1000).build
                                          

  processA will be created with buffer size limit equals 5000

  processB will be created with buffer size limit equals 1000

  Example 3:

    override val config: ParConfig = ParConfig(5000, SchedulerConfig.default)
  
    val processA = Process[IO](_ => {
      case _ => unit
    })
  
    val processB = Process.builder[IO](_ => {
      case _ => unit
    }).bufferSize(-1).build
                                          

  processA will be created with buffer size limit equals 5000

  processB will be created with unbounded buffer

  You should set bufferSize to a value that would match the expected workload. For example, if you are going to send 1M events within the same flow it's recommended to set bufferSize to 1M. However, it depends on how fast your consumer processes and amount of available memory, if that's possible to keep some amount of events in memory - go for it, if not - you will probably need to reconsider your design decisions. In a case the process event queue is full all events will be redirected to EventLog (see the corresponding section).

• schedulerConfig:
  • numberOfWorkers - number of workers; default = availableProcessors

Safty properties:

• It's guaranteed that events will be delivered to a process in a strictly synchronous request-reply dialog, i.e. a process will receive a new event iff it completed processing the current one.
• All events delivered in send order

Liveness properties:

• Sent events eventually delivered
• A sender eventually receives a response

Messaging - #back to top

A distributed system consists of a set of processors that are connected by a communication network. The communication network provides the facility of information exchange among processors.

Parapet provides a messaging module that consists of an API - set of events that can be used to implement basic communication protocols and messaging abstractions, e.g.: client/server. Implementations based on concrete libraries reside in corresponding subprojects.

In order to start using messaging module you need to add two dependencies: messaging-api and messaging-{specific_library}.

By default, Parapet provides implementations of basic messaging components based on ZMQ library.

Using Parapet with ZMQ

Add the following libraries to your project:

                                    libraryDependencies += "io.parapet" %% "messaging-api" % version
                                    libraryDependencies += "io.parapet" %% "messaging-zmq" % version
                                

Sync client and sync server

ZmqSyncClient - based on REQ socket type

ZmqSyncServer - based on REP socket type

Example:

import java.net.ServerSocket

import cats.effect.IO
import io.parapet.CatsApp
import io.parapet.core.Event.Start
import io.parapet.core.{Encoder, Event, Process}
import io.parapet.messaging.api.MessagingApi.{Failure, Request, Success}
import io.parapet.messaging.{ZmqSyncClient, ZmqSyncServer}

object SyncClientSyncServer extends CatsApp {

  import dsl._

  // Application API
  case class TestRequest(data: Any) extends Event
  case class TestResponse(data: Any) extends Event

  // Required for encoding/decoding events
  private val encoder =
    Encoder.json(
      List(
        // Messaging Api
        classOf[Request],
        classOf[Success],
        classOf[Failure],
        // Application Api
        classOf[TestRequest],
        classOf[TestResponse]
      )
    )

  override def processes: IO[Seq[Process[IO]]] = IO {
    val port = new ServerSocket(0).getLocalPort

    val echoService = Process[IO](_ => {
      case TestRequest(id) => withSender(sender => TestResponse(s"echo: $id") ~> sender)
    })

    val zmqServer = ZmqSyncServer[IO](s"tcp://*:$port", echoService.ref, encoder)
    val zmqClient = ZmqSyncClient[IO](s"tcp://localhost:$port", encoder)

    val client = Process[IO](_ => {
      case Start => Request(TestRequest("hello")) ~> zmqClient
      case Success(TestResponse(data)) => eval(println(data))
    })

    Seq(zmqClient, zmqServer, echoService, client)
  }

}

echo: hello

Async client and async server

ZmqAsyncClient - based on DEALER socket type

ZmqAsyncServer - based on ROUTER socket type

Example:

import java.net.ServerSocket

import cats.effect.IO
import io.parapet.CatsApp
import io.parapet.core.Event.Start
import io.parapet.core.{Encoder, Event, Process, ProcessRef}
import io.parapet.messaging.api.MessagingApi.{Failure, Request, Success}
import io.parapet.messaging.api.ServerAPI.Envelope
import io.parapet.messaging.{ZmqAsyncClient, ZmqAsyncServer}


object AsyncClientAsyncServer extends CatsApp {

  import dsl._

  // Application API
  case class TestRequest(data: Any) extends Event
  case class TestResponse(data: Any) extends Event

  private val encoder = Encoder.json(
    List(
      // Messaging Api
      classOf[Request],
      classOf[Success],
      classOf[Failure],
      // Application Api
      classOf[TestRequest],
      classOf[TestResponse])
  )

  override def processes: IO[Seq[Process[IO]]] = IO.suspend {
    val port = new ServerSocket(0).getLocalPort
    val nClients = 5
    val nClientWorkers = 5
    val nServerWorkers = 5
    for {
      echoService <- IO(Process[IO](_ => {
        case Envelope(requestId, TestRequest(id)) =>
          fork(withSender(sender => Envelope(requestId, TestResponse(id)) ~> sender))
      }))
      zmqClient <- ZmqAsyncClient[IO](s"tcp://localhost:$port", encoder, nClientWorkers)
      zmqServer <- IO(ZmqAsyncServer[IO](s"tcp://*:$port", echoService.ref, encoder, nServerWorkers))
      clients <- IO((0 until nClients).map(createClient(_, zmqClient.ref)))
    } yield Seq(echoService, zmqClient, zmqServer) ++ clients

  }

  def createClient(id: Int, zmqClient: ProcessRef): Process[IO] = Process[IO](_ => {
    case Start => Request(TestRequest(id)) ~> zmqClient
    case Success(TestResponse(data)) => eval(println(s"client-$id received $data"))
  })
}

client-2 received 2
client-3 received 3
client-0 received 0
client-4 received 4
client-1 received 1

The Freelance Protocol

The Freelance Protocol (FLP) defines brokerless reliable request-reply dialogs across an N-to-N network of clients and servers. Protocol specification

Example:

import java.net.ServerSocket

import cats.effect.{Concurrent, IO}
import io.parapet.CatsApp
import io.parapet.core.Dsl.DslF
import io.parapet.core.Event.{Start, Stop}
import io.parapet.core.{Channel, Encoder, Event, Process, ProcessRef}
import io.parapet.messaging.api.{FLProtocolApi, HeartbeatAPI, MessagingApi, ServerAPI}
import io.parapet.messaging.{FLProtocol, ZmqAsyncServer}

import scala.concurrent.duration._

object FLProtocolExample extends CatsApp {

  import dsl._

  // Application API
  case class TestRequest(data: Any) extends Event
  case class TestResponse(data: Any) extends Event

  val encoder = Encoder.json(List(

    // Messaging Api
    classOf[MessagingApi.Request],
    classOf[MessagingApi.Success],
    classOf[MessagingApi.Failure],

    // Application Api
    classOf[TestRequest],
    classOf[TestResponse],

    // Heartbeat Api
    HeartbeatAPI.Ping.getClass,
    HeartbeatAPI.Pong.getClass,

    // FL Api
    classOf[FLProtocolApi.Connect],

    // Server Api
    classOf[ServerAPI.Envelope]
  ))


  class FLTestClient[F[_] : Concurrent](flprotocol: ProcessRef, nRequests: Int,
                                        servers: List[String]) extends Process[F] {

    import dsl._

    val ch = new Channel[F]()

    override def handle: Receive = {
      case Start =>
        register(ref, ch) ++
          servers.map(endpoint => FLProtocolApi.Connect(endpoint)) ~> flprotocol ++
          delay(1.second, generateRequests(nRequests, ch))
      case Stop => unit
    }

    def generateRequests(n: Int, ch: Channel[F]): DslF[F, Unit] = {
      (0 until n).map { i =>
        ch.send(MessagingApi.Request(TestRequest(i.toString)), flprotocol, {
          case scala.util.Success(res) => eval(println(res))
          case scala.util.Failure(err) => eval(println(err))
        })
      }.fold(unit)(_ ++ _)
    }
  }

  override def processes: IO[Seq[Process[IO]]] = IO.suspend {
    val availableServerPort = new ServerSocket(0).getLocalPort
    val nRequests = 5
    val flprotocol = new FLProtocol[IO](encoder)

    val service = Process[IO](_ => {
      case ServerAPI.Envelope(id, TestRequest(body)) =>
        withSender(sender => {
          ServerAPI.Envelope(id, TestResponse("server-" + body)) ~> sender
        })
    })

    val client = new FLTestClient[IO](flprotocol.ref, nRequests,
      List("tcp://localhost:4444", // unavailable server
        "tcp://localhost:5555", // unavailable server
        s"tcp://localhost:$availableServerPort"))

    val server = ZmqAsyncServer[IO](s"tcp://*:$availableServerPort",
      service.ref, encoder, numOfWorkers = 5, s"tcp://localhost:$availableServerPort")
    IO(Seq(service, flprotocol, client, server))
  }
}

Failure(request expired,104)
Failure(request expired,104)
Failure(request expired,104)
Success(TestResponse(server-3))
Success(TestResponse(server-4))

Work distribution - #back to top

Failure detection - #back to top

Consensus and agreement - #back to top

Mutual exclusion - #back to top

Snapshots - #back to top

Deadlock detection - #back to top

Election - #back to top

Akka - #back to top

• If you are already using any Effect System library which is supported in Parapet, e.g. Cats Effect
• You'd like to have a possibility of switching Effect System libraries
• You'd like to have access to internal components and distributed algorithms and use them in your code
• You prefer functional programming over imperative

Messages: 1000000
Consumers: 1
Producers: 1
                                

The project in its early stage and many things are subject to change. Now is a good time to join! If you want to become a contributor please send an email or text in gitter channel.

If you'd like to donate in order to help with ongoing development and maintenance:

   Copyright [2019] The Parapet Project Developers

   Licensed 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