Akka scala Event автобус с различными классификаторами в зависимости от абонента

Question

Я изучаю Akka EventBus, чтобы проверить, может ли он решить одну из моих проблем с дизайном, и я до сих пор не знаю. Проблема заключается в следующем.

упростить, у меня есть:

case class Request(requesterId: String, operation: String, header: RequestHeader) 
case class Response(requesterId: String, operation: String, header: ResponseHeader) 

У меня есть несколько актеров с различными функциями, и я хочу некоторые актер подписаться на Response в зависимости от requesterId, некоторые другие в зависимости от operation. Есть ли способ достичь этого легко, с EventBus и классификаторами?

Спасибо, Joel

Answer 1

Конечно, это называется LookupEventBus. Вы реализуете свой собственный автобус, расширяя его и извлечь requesterId в методе classify, как это:

class LookupBusImpl extends EventBus with LookupClassification { 
    type Event = HasRequesterId // I made up a super type for you here 
    type Classifier = String 
    type Subscriber = ActorRef 

    override def classify(event: HasRequesterId): String = event.requesterId 

Тогда вы бы подписаться на данной requesterId, например, так:

lookupBus.subscribe(actorRef, "requester-100") 

И этот актер затем будут получать только сообщения, которые были классифицированы как requester-100.

Answer 2

Я согласен с Konrad, что вы должны внедрить новые шины LookupClassification, чтобы решить вашу проблему. Я думаю, что проще всего иметь два отдельных экземпляра этих автобусов: один, который классифицирует by requesterId, а другой - с помощью операции. Некоторые из базовой настройки работы для такого подхода было бы:

//Singleton to hold the instances of each stream type 
object ResponseEventStream{ 
    val RequestorIdStream = new RequestorIdResponseEventStream 
    val OperationStream = new OperationResponseEventStream 
} 

//Common functionality for the two different types of streams 
trait ResponseEventStream extends ActorEventBus with LookupClassification{ 
    import ResponseEventStream._ 
    type Event = Response 
    type Classifier = String 
    protected def mapSize = 128 
    protected def publish(resp:Response, subscriber: ActorRef) = { 
    if (subscriber.isTerminated) unsubscribe(subscriber) 
    else subscriber ! resp 
    } 
} 

//Concrete impl that uses requesterId to classify 
class RequestorIdResponseEventStream extends ResponseEventStream{ 
    protected def classify(resp:Response) = resp.requesterId 
} 

//Concrete impl that uses operation to classify 
class OperationResponseEventStream extends ResponseEventStream{ 
    protected def classify(resp:Response) = resp.operation 
} 

//Trait to mix into classes that need to publish or subscribe to response events 
//Has helper methods to simplify interaction with the two distinct streams 
trait ResponseEventing{ 
    import ResponseEventStream._ 

    def publishResponse(resp:Response){ 
    RequestorIdStream.publish(resp) 
    OperationStream.publish(resp) 
    } 

    def subscribeByRequestId(requestId:String, ref:ActorRef){ 
    RequestorIdStream.subscribe(ref, requestId) 
    } 

    def subscribeByOperartion(op:String, ref:ActorRef){ 
    OperationStream.subscribe(ref, op) 
    } 
} 

Тогда вам просто нужно смешать эту ResponseEventing черту в актеров, которые должны либо публиковать Response события или актеров, которые должны подписаться на них. Актеры, которые публикуют, будут звонить publishResponse, и актеры, которым необходимо подписаться, будут звонить subscribeXXX в зависимости от того, какую категорию они интересуют.