path: root/cpp/test/IceStorm/federation2/run.py

#!/usr/bin/env python
# **********************************************************************
#
# Copyright (c) 2003-2007 ZeroC, Inc. All rights reserved.
#
# This copy of Ice is licensed to you under the terms described in the
# ICE_LICENSE file included in this distribution.
#
# **********************************************************************

import os, sys, time, threading, re

for toplevel in [".", "..", "../..", "../../..", "../../../.."]:
    toplevel = os.path.normpath(toplevel)
    if os.path.exists(os.path.join(toplevel, "config", "TestUtil.py")):
        break
else:
    raise "can't find toplevel directory!"

sys.path.append(os.path.join(toplevel, "config"))
import TestUtil
TestUtil.processCmdLine()

name = os.path.join("IceStorm", "federation2")
testdir = os.path.dirname(os.path.abspath(__file__))

import IceStormUtil

iceBox = TestUtil.getIceBox(testdir)
iceBoxAdmin = os.path.join(TestUtil.getBinDir(testdir), "iceboxadmin")
iceStormAdmin = os.path.join(TestUtil.getBinDir(testdir), "icestormadmin")

publisher = os.path.join(testdir, "publisher")
subscriber = os.path.join(testdir, "subscriber")

def admin(ref, command):
    pipe = TestUtil.startClient(iceStormAdmin, ref + r' -e "%s"' % command)
    all = ""
    while True:
        line = pipe.readline();
        if not line:
            break
        all = all + line
    status = TestUtil.closePipe(pipe)
    if status:
        TestUtil.killServers()
        sys.exit(1)
    return all

def printOutput(pipe):
    try:
        while True:
            line = pipe.readline()
            if not line:
                break
            print line,
            sys.stdout.flush()
    except IOError:
        pass

def runPublisher(icestorm1, opt = ""):
    publisherPipe = TestUtil.startClient(publisher, opt + icestorm1.reference())

    printOutput(publisherPipe)

    publisherStatus = TestUtil.closePipe(publisherPipe)
    if publisherStatus:
        TestUtil.killServers()
        sys.exit(1)

def doTest(icestorm1, icestorm2, batch, subscriberRef = None):
    if batch:
        name = "batch subscriber"
        batchOptions = " -b"
    else:
        name = "subscriber"
        batchOptions = ""

    if subscriberRef == None:
        subscriberRef = icestorm2.reference()

    subscriberPipe = TestUtil.startServer(subscriber, batchOptions + subscriberRef)
    TestUtil.getServerPid(subscriberPipe)
    TestUtil.getAdapterReady(subscriberPipe)

    #
    # Start the publisher. This should publish events which eventually
    # causes subscriber to terminate.
    #
    runPublisher(icestorm1)

    subscriberStatus = TestUtil.specificServerStatus(subscriberPipe, 30)
    return subscriberStatus

#
# Test #1:
#
# create a cross service link fed1->fed2 and ensure the events are
# published between them correctly.
#
def runtest(type, **args):
    icestorm1 = IceStormUtil.init(toplevel, testdir, type, additional = '--IceStorm.Discard.Interval=2',
                                  dbDir = "db", instanceName = "TestIceStorm1", port = 12000, **args)
    icestorm1.start()
    icestorm2 = IceStormUtil.init(toplevel, testdir, type, additional = '--IceStorm.Discard.Interval=2',
                                  dbDir = "db2", instanceName = "TestIceStorm2", port = 12500, **args)
    icestorm2.start()

    adminIceStormReference = ' --IceStormAdmin.TopicManager.Proxy="%s" --IceStormAdmin.TopicManager.Proxy2="%s"' % (
        icestorm1.proxy(), icestorm2.proxy())

    print "setting up the topics...",
    sys.stdout.flush()
    admin(adminIceStormReference, "create TestIceStorm1/fed1 TestIceStorm2/fed1; link TestIceStorm1/fed1 TestIceStorm2/fed1")
    print "ok"

    #
    # Test oneway subscribers.
    #
    print "testing federation with oneway subscribers...",
    sys.stdout.flush()
    onewayStatus = doTest(icestorm1, icestorm2, 0)
    print "ok"

    #
    # Test batch oneway subscribers.
    #
    print "testing federation with batch subscribers...",
    sys.stdout.flush()
    batchStatus = doTest(icestorm1, icestorm2, 1)
    print "ok"

    if onewayStatus or batchStatus:
        TestUtil.killServers()
        sys.exit(1)

    #
    # Test #2:
    #
    # Stop and restart the service and repeat the test. This ensures that
    # the database is correct.
    #
    print "restarting services to ensure that the database content is preserved...",
    sys.stdout.flush()

    #
    # Shutdown icestorm.
    #
    icestorm1.stop()
    icestorm2.stop()

    icestorm1.start(echo=False)
    icestorm2.start(echo=False)
    print "ok"

    #
    # Test oneway subscribers.
    #
    print "retesting federation with oneway subscribers... ",
    sys.stdout.flush()
    onewayStatus = doTest(icestorm1, icestorm2, 0)
    print "ok"

    #
    # Test batch oneway subscribers.
    #
    print "retesting federation with batch subscribers... ",
    sys.stdout.flush()
    batchStatus = doTest(icestorm1, icestorm2, 1)
    print "ok"

    if onewayStatus or batchStatus:
        TestUtil.killServers()
        sys.exit(1)

    #
    # Shutdown icestorm.
    #
    icestorm1.stop()
    icestorm2.stop()

    #
    # This is used by the below test to confirm that the link warning is
    # emitted. This class conforms with the TestUtil.ReaderThread protocol.
    #
    class ExpectorThread(threading.Thread):
        def __init__(self, pipe):
            self.mutex = threading.Lock()
            self.pipe = pipe
            # Suppress "adapter ready" messages. Under windows the eol isn't \n.
            self.re = [ [ re.compile(" ready\r?\n$"), 0 ] ]
            threading.Thread.__init__(self)

        def run(self):
            try:
                while 1:
                    line = self.pipe.readline()
                    if not line: break
                    found = False
                    self.mutex.acquire()
                    for item in self.re:
                        if item[0].search(line):
                            found = True
                            item[1] = item[1] + 1
                            break
                    self.mutex.release()
                    if not found:
                        print line,
            except IOError:
                pass

            self.status = TestUtil.closePipe(self.pipe)

        # To comply with the ReaderThread protocol.
        def getPipe(self):
            return self.pipe

        # To comply with the ReaderThread protocol.
        def getStatus(self):
            return self.status

        def matches(self, index):
            self.mutex.acquire()
            m = self.re[index][1]
            self.mutex.release()
            return m

        def expect(self, r):
            self.mutex.acquire()
            self.re.append([r, 0])
            l = len(self.re)-1
            self.mutex.release()
            return l

    #
    # Test #3:
    #
    # Restart the first server and publish some events. Attach a
    # subscriber to the channel and make sure the events are received.
    #
    # Then re-start the linked downstream server and publish the events.
    # Ensure they are received by the linked server.
    #
    if type != "replicated":
        print "restarting only one IceStorm server...",
        sys.stdout.flush()
        pipe = icestorm1.start(echo=False, createThread = False)
        expectorThread = ExpectorThread(pipe)
        expectorThread.start()
        global serverThreads

        TestUtil.serverThreads.append(expectorThread)
        index = expectorThread.expect(re.compile("topic.fed1.*subscriber offline"))
        expectorThread.expect(re.compile("connection refused"))
        print "ok"

        #
        # Test oneway subscribers.
        #
        print "testing that the federation link reports an error...",
        sys.stdout.flush()
        onewayStatus = doTest(icestorm1, icestorm2, 0, icestorm1.reference())

	# Give some time for the output to be sent.
	time.sleep(2)

        if onewayStatus or expectorThread.matches(index) != 1:
            TestUtil.killServers()
            sys.exit(1)
        print "ok"

        print "starting downstream icestorm server...",
        sys.stdout.flush()
        icestorm2.start(echo=False)
        print "ok"

        #
        # Need to sleep for at least the discard interval.
        #
        time.sleep(3)

        #
        # Test oneway subscribers.
        #
        print "testing link is reestablished...",
        sys.stdout.flush()
        onewayStatus = doTest(icestorm1, icestorm2, 0)
        print "ok"

        if onewayStatus or expectorThread.matches(index) != 1:
            TestUtil.killServers()
            sys.exit(1)

        icestorm1.stop()
        icestorm2.stop()

    #
    # Test #4:
    #
    # Trash the TestIceStorm2 database. Then restart the servers and
    # verify that the link is removed.
    #
    print "destroying the downstream IceStorm service database...",
    sys.stdout.flush()
    icestorm2.clean()

    print "ok"

    print "restarting IceStorm servers...",
    sys.stdout.flush()
    icestorm1.start(echo = False)
    icestorm2.start(echo = False)
    print "ok"

    print "checking link still exists...",
    line = admin(adminIceStormReference, "links TestIceStorm1")
    if not re.compile("fed1 with cost 0").search(line):
        print line
        TestUtil.killServers()
        sys.exit(1)
    print "ok"

    print "publishing some events...",
    sys.stdout.flush()
    # The publisher must be run twice because all the events can be
    # sent out in one batch to the linked subscriber which means that
    # the link is not reaped until the next batch is
    # sent. Furthermore, with a replicated IceStorm both sets of
    # events must be set to the same replica.
    runPublisher(icestorm1, opt = " --count 2")
    print "ok"

    # Give the unsubscription time to propagate.
    time.sleep(1)

    # Verify that the link has disappeared.
    print "verifying that the link has been destroyed...",
    sys.stdout.flush()
    line = admin(adminIceStormReference, "links TestIceStorm1")
    if len(line) > 0:
        print line
        TestUtil.killServers()
        sys.exit(1)
    print "ok"

    #
    # Destroy the remaining topic.
    #
    print "destroying topics...",
    admin(adminIceStormReference, "destroy TestIceStorm1/fed1")
    print "ok"

    #
    # Shutdown icestorm.
    #
    print "shutting down icestorm services...",
    sys.stdout.flush()
    icestorm1.stop()
    icestorm2.stop()
    print "ok"

runtest("persistent")
runtest("replicated", replicatedPublisher = False)
runtest("replicated", replicatedPublisher = True)

sys.exit(0)