Re: [akka-user] DistributedPubSubExtension, subscriber doesn't receive published message

'Konstantinos Kougios' via Akka User List Sun, 28 Jun 2015 23:07:59 -0700

Ok thanks, updated akka and waiting for all mediators to be aware of thetopic


On 26/06/15 08:03, Patrik Nordwall wrote:

On Fri, Jun 26, 2015 at 7:42 AM, 'Konstantinos Kougios' via Akka UserList <akka-user@googlegroups.com <mailto:akka-user@googlegroups.com>>wrote:


    Thanks, gossip-interval works, but also tried 2.4-M1. I get a
    compilation error that DistributedPubSubExtension is not found and
    can't find DistributedPubSubExtension class in those deps:

    http://search.maven.org/#search|ga|1|c%3A%22DistributedPubSubExtension%22

It has moved, see migration guide:http://doc.akka.io/docs/akka/2.4-M1/project/migration-guide-2.3.x-2.4.x.html#Cluster_tools_moved_to_separate_module

and alsohttp://doc.akka.io/docs/akka/2.4-M1/project/migration-guide-2.3.x-2.4.x.html#DistributedPubSub_construction




    On 25/06/15 13:04, Patrik Nordwall wrote:

    Reduce

    akka.contrib.cluster.pub-sub.gossip-interval



    In 2.4-M1 there is a GetTopics message that could be used
    together with awaitAssert to query it until it has the expected
    topics.

    On Thu, Jun 25, 2015 at 8:16 AM, 'Konstantinos Kougios' via Akka
    User List <akka-user@googlegroups.com
    <mailto:akka-user@googlegroups.com>> wrote:

        Ok, thanks, a Thread.sleep(2000) sorted it out. But now we go
        back to the situation similar to my other post. How do I
        kickstart that consistency. Currently my test waits all
        clusters to join and then sleeps for 2 secs for the
        DistributedPubSub to become consistent. The good news is all
        my tests now pass :)

        awaitCond(
            cluster1.state.members.size ==3&&cluster2.state.members.size 
==3&&driverCluster.state.members.size ==3,
            60seconds
        )

        Thread.sleep(2000)


        On 24/06/15 13:03, Patrik Nordwall wrote:



        On Tue, Jun 23, 2015 at 8:06 AM, Kostas kougios
        <kostas.koug...@googlemail.com
        <mailto:kostas.koug...@googlemail.com>> wrote:

            Hi, using akka cluster and DistributedPubSubExtension, I
            am creating a topic.

            The receiving actor (driver) subscribes to the topic:

            private valmediator= DistributedPubSubExtension(system).mediator
            mediator!Subscribe(Subscriptions.IndexActorAvailability,self)


            override defreceive= {
                caseSubscribeAck(subscribe) =>
                   context becomeready
            ...


            The publishing actor (server) publishes a msg during
            it's startup:

            private valmediator= 
DistributedPubSubExtension(context.system).mediator
               
mediator!Publish(Subscriptions.IndexActorAvailability,IAmAvailable(index,self))


            But the driver actor never receives it.

            Some notes:

            1. "context become ready" is certainly executed before
            the publishing actor publishes the msg (I've debugged it)


        What is your definition of before, here?
        The registry of subscribers is eventually consistent, i.e.
        changes are not immediately visible at other nodes, but
        typically they will be fully replicated to all other nodes
        after a few seconds.

        /Patrik


            2. The actors context.system are different (akka -
            cluster) but the rest of the cluster communicates fine.

            3. all run under a testcase in the same jvm

            Here is my config:

            akka{
               log-dead-letters-during-shutdown= off
               loglevel="INFO"
               cluster{
                 gossip-interval= 100 ms
                 leader-actions-interval= 100 ms
                 seed-nodes= [
                   "akka.tcp://testSystem@127.0.0.1:2700  
<http://testSystem@127.0.0.1:2700>",
                   "akka.tcp://testSystem@127.0.0.1:2701  
<http://testSystem@127.0.0.1:2701>"
                 ]
               }
               extensions= ["akka.contrib.pattern.DistributedPubSubExtension"]
            }

            server1{
               akka{
                 remote{
                   netty.tcp{
                     port=2700
                   }
                 }
                 cluster{
                   roles= ["index-server"]
                 }
               }
            }

            server2{
               akka{
                 remote{
                   netty.tcp{
                     port=2701
                   }
                 }
                 cluster{
                   roles= ["index-server"]
                 }
               }
            }

            driver{
               akka{
                 remote{
                   netty.tcp{
                     port=2702
                   }
                 }
                 cluster{
                   roles= ["driver"]
                 }
               }
            }