PyRQ.Iface.PyRQIface.PyRQIface.create

All Samples(7) | Call(7) | Derive(0) | Import(0)

@summary: Open a connection to the RRQ and ask it to create a new 'Q'.

src/p/y/PyRQ-0.4.1/test/RRQReader/TestQueueBlocking.py PyRQ(Download)

        """
        iface = PyRQIface(quiet=self.quiet, ref="creator-testDelayedGetWithTimeoutThenClosed")
        namespace = iface.create(maxsize=maxsize)
        self._closeQueue(namespace, delay=2, name="closer-testDelayedGetThenClosed")
        iface = PyRQIface(namespace=namespace, quiet=self.quiet, ref="testDelayedGetThenClosed")

        """
        iface = PyRQIface(quiet=self.quiet, ref="creator-testDelayedGetWithTimeoutThenClosed")
        namespace = iface.create(maxsize=maxsize)
        self._closeQueue(namespace, delay=1, name="closer-testDelayedGetWithTimeoutThenClosed")
        iface = PyRQIface(namespace=namespace, quiet=self.quiet, ref="testDelayedGetWithTimeoutThenClosed")

    def testDelayedPutThenClosed(self, maxsize=1):
        #    Put 2 items, second one should block forever, then we close the q.
        iface = PyRQIface(quiet=self.quiet, ref="creator-testDelayedPutThenClosed")
        namespace = iface.create(maxsize=maxsize)
        iface = PyRQIface(namespace=namespace, quiet=self.quiet, ref="testDelayedPutThenClosed")

    def testDelayedWithTimeoutPutThenClosed(self, maxsize=1):
        #    Put 2 items, second one should block forever, then we close the q.
        iface = PyRQIface(quiet=self.quiet, ref="creator-testDelayedWithTimeoutPutThenClosed")
        namespace = iface.create(maxsize=maxsize)
        iface = PyRQIface(namespace=namespace, quiet=self.quiet, ref="testDelayedWithTimeoutPutThenClosed")

    def testPutOnFullQueueWithTimeout(self, maxsize=1):
        #    Put 2 items, second one should block forever, then we close the q.
        iface = PyRQIface(quiet=self.quiet, ref="creator-testDelayedWithTimeoutPutThenClosed")
        namespace = iface.create(maxsize=maxsize)
        iface = PyRQIface(namespace=namespace, quiet=self.quiet, ref="testDelayedWithTimeoutPutThenClosed")

src/p/y/PyRQ-0.4.1/test/RRQReader/TestRRQReader.py PyRQ(Download)

    def _createInterface(self):
        namespace = self.iface.create()
        self.iface = PyRQIface(namespace=namespace, ref="test", quiet=self.quiet, loggingModule=testLoggingModule)
        self.namespaces.append(namespace)
        return namespace

        try:
            try:
                iface.create(timeout=timeout)
            except PyRQError, _e:
                assert True