1
\$\begingroup\$

I extracted a small part of my project, it is a workflow engine based on vertx rx-java and Kotlin. This part represent the code use to manage Bpmn Service task. The workflow is a stack of ServiceTask calls...

Does anyone have 5 minutes for a quick review of this?

GitHub project

abstract class ServiceTask: Handler < Message < MessageBody >> , Loggify {
  val properties: MutableList < Property > = mutableListOf()
  lateinit
  var node: Node
  lateinit
  var eventBus: EventBus
  lateinit
  var vertx: Vertx
  lateinit
  var eventName: String

  companion object {
    val EVENT_ENDED = "EVENT_ENDED"
  }

  override fun handle(message: Message < MessageBody > ) {
    logger.info("Handle event message ${message.body()} for event: $eventName for nodeId ${node.id}")
    val messageBody = message.body()
    safeHandle(message,
        messageBody)
      .subscribe({
        serviceTaskResult - >
        send(serviceTaskResult.eventName, serviceTaskResult.messageBody)
      }, {
        if (it != null) throw it
        else {
          message.reply(null)
          ko()
        }
      }, {
        message.reply(null)
        ok(messageBody).subscribeAndReply(message)
      })
  }

  abstract fun safeHandle(message: Message < MessageBody > , messageBody: MessageBody): Observable < ServiceTaskResult >

    abstract fun checkValidity(): Boolean

  abstract fun init()

  /**
   * Send event to other nodes
   */
  private fun < T > send(eventName: String, messageBody: T): rx.Observable < Message < T >> {
    return eventBus.rxSend < T > (eventName, messageBody)
      .toObservable()
  }

  /**
   * Call when Service Task job is done
   * Will getValidNextEvent all nextLink to see if there is a valid condition
   */
  private fun ok(messageBody: MessageBody): Observable < ServiceTaskResult > {

    if (logger.isDebugEnabled) {
      logger.debug("Ok called for node ${node.id}")
    }

    send < JsonObject > (EVENT_ENDED, JsonObject().put("eventName", eventName).put("nodeId", node.id)).subscribeAndLog()

    return FlowOrchestrator.tryToDoNode(node = node, messageBody = messageBody)
  }

  /**
   * Stop the call chains here
   */
  private fun ko() {

    if (logger.isDebugEnabled) {
      logger.debug("Dropped for node ${node.id}")
    }
    send < JsonObject > (EVENT_ENDED, JsonObject().put("eventName", eventName).put("nodeId", node.id)).subscribeAndLog()
  }

  /**
   * Register inside eventBus
   */
  fun register(eventBus: EventBus, flowName: String, eventName: String, node: Node) {
    logger.info("Try to Register ${this::class.simpleName} for event $eventName")
    this.node = node
    this.eventBus = eventBus
    this.eventName = eventName
    init()
    checkValidity()
    EventBusConsumerManager.register(flowName, eventName, this)
  }


  private fun Observable < ServiceTaskResult > .subscribeAndReply(message: Message < MessageBody > ) {
    this.subscribe(
      reply(message),
      fail(message),
      complete())
  }

  private fun reply(message: Message < MessageBody > ): (ServiceTaskResult) - > Unit = {
    message.reply("")
  }

  private fun fail(message: Message < MessageBody > ): (Throwable) - > Unit = {
    message.fail(0, it.message)
  }

  private fun complete() = {}

}


------------------------------------------------------


    object FlowOrchestrator : Loggify {

        fun getValidNextEvent(node: Node, messageBody: MessageBody): List<String> {
            if (logger.isDebugEnabled)
                logger.debug("${this.hashCode()}: Try to check condition on node ${node.id}")
            return AllLinksConditionChecker.getValidNextEvent(node, messageBody)
        }

        fun tryToDoNode(validEvents: MutableList<String> = mutableListOf(),
                        node: Node, messageBody: MessageBody)
                : Observable<ServiceTaskResult> {
            if (validEvents.isEmpty())
                validEvents.addAll(FlowOrchestrator.getValidNextEvent(node, messageBody))
            return if (validEvents.isNotEmpty()) {
                validEvents.toObservable().flatMap { notifyNextNode(node, it, messageBody) }
            } else {
                Observable.empty()
            }
        }

        private fun notifyNextNode(node: Node, it: String, messageBody: MessageBody): Observable<ServiceTaskResult>? {
            if (logger.isDebugEnabled)
                logger.debug("${this.hashCode()}: The condition is valid for node ${node.id}, sending event $it with messageBody $messageBody")
            return Observable.just(ServiceTaskResult(it, messageBody))
        }

        fun replayCurrentNode(node: Node, messageBody: MessageBody): Observable<ServiceTaskResult>? {
            val eventName = node.previousLinks.first().eventName
            if (logger.isDebugEnabled)
                logger.debug("${this.hashCode()}: The condition is not valid for node ${node.id}, sending event $eventName with messageBody $messageBody")
            return Observable.just(ServiceTaskResult(eventName, messageBody))
        }
    }

 class HumanServiceTask : ServiceTask(), Loggify {
        private lateinit var humanNode: HumanNode

        override fun init() {
            if (node !is HumanNode) throw HumanServiceTaskMustBeLinkToHumanNodeException(node.id)
            humanNode = node as HumanNode
        }

        override fun checkValidity(): Boolean {
            if (humanNode.responseList.isEmpty()) throw HumanServiceTaskMustHaveTxtOrMediaException(humanNode.responseList.all().joinToString(","))
            return true
        }

        override fun safeHandle(message: Message<MessageBody>, messageBody: MessageBody): Observable<ServiceTaskResult> {
            messageBody.updateWithNode(node)
            return Observable.just(ServiceTaskResult(EventBusEvent.SET_MSG.name, messageBody))
        }

        class HumanServiceTaskMustBeLinkToHumanNodeException(id: String) : RuntimeException("Human service task must be link to human node $id")
        class HumanServiceTaskMustHaveTxtOrMediaException(txt: String) : RuntimeException("Human service task must have txt or media $txt")

    }

enter image description here

The goal is to validate my use of rx, it’s the first time I use it

Thanks

\$\endgroup\$
  • 1
    \$\begingroup\$ Welcome to Code Review. Can you give a little more detail about what your code does, rather than just specifying what libraries you use? (that's what the tags are for) It's not immediately clear with only the code, it would help us help you better. \$\endgroup\$ – Phrancis Oct 22 '17 at 14:01
  • \$\begingroup\$ Great, thanks! I hope you get some great answers. \$\endgroup\$ – Phrancis Oct 22 '17 at 14:15
  • \$\begingroup\$ What's it supposed to do? \$\endgroup\$ – Mast Oct 22 '17 at 14:19
  • \$\begingroup\$ THe vertx eventBus call the handle method that call safeHandle method inside the child class. THe FlowOrchestrator test the next node and if the node is ok it send event to the eventbus \$\endgroup\$ – fstn Oct 22 '17 at 14:23
  • \$\begingroup\$ Yea, but why? What problem does it solve? \$\endgroup\$ – Mast Oct 22 '17 at 14:24

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.