def set_snapshot(xmin, snapshot_id):
    global current_xmin_snapshot_id
    if current_xmin_snapshot_id == (xmin, snapshot_id):
        return
    clear_snapshot()
    current_xmin_snapshot_id = (xmin, snapshot_id)

    while True:
        txn = transaction.begin()
        txn.doom()
        if snapshot_id is not None:
            txn.setExtendedInfo('snapshot_id', snapshot_id)
        session = app.registry[DBSESSION]()
        connection = session.connection()
        db_xmin = connection.execute(
            "SELECT txid_snapshot_xmin(txid_current_snapshot());").scalar()
        if db_xmin >= xmin:
            break
        transaction.abort()
        log.info('Waiting for xmin %r to reach %r', db_xmin, xmin)
        time.sleep(0.1)

    registry = app.registry
    request = app.request_factory.blank('/_indexing_pool')
    request.registry = registry
    request.datastore = 'database'
    apply_request_extensions(request)
    request.invoke_subrequest = app.invoke_subrequest
    request.root = app.root_factory(request)
    request._stats = {}
    manager.push({'request': request, 'registry': registry})

    def __call__(self, environ, start_response):
        """
        Accept ``environ`` and ``start_response``; create a
        :term:`request` and route the request to a :app:`Pyramid`
        view based on introspection of :term:`view configuration`
        within the application registry; call ``start_response`` and
        return an iterable.
        """
        registry = self.registry
        has_listeners = self.registry.has_listeners
        notify = self.registry.notify
        request = self.request_factory(environ)
        threadlocals = {"registry": registry, "request": request}
        manager = self.threadlocal_manager
        manager.push(threadlocals)
        request.registry = registry
        try:

            try:
                response = self.handle_request(request)
                has_listeners and notify(NewResponse(request, response))

                if request.response_callbacks:
                    request._process_response_callbacks(response)

                return response(request.environ, start_response)

            finally:
                if request.finished_callbacks:
                    request._process_finished_callbacks()

        finally:
            manager.pop()

    def invoke_request(self, request,
                       _use_tweens=True, _apply_extensions=False):
        registry = self.registry
        has_listeners = self.registry.has_listeners
        notify = self.registry.notify
        threadlocals = {'registry': registry, 'request': request}
        manager = self.threadlocal_manager
        manager.push(threadlocals)

        if _use_tweens:
            handle_request = self.handle_request
        else:
            handle_request = self.orig_handle_request

        try:

            try:
                extensions = self.request_extensions
                if _apply_extensions and extensions is not None:
                    apply_request_extensions(request, extensions=extensions)
                response = handle_request(request)

                if request.response_callbacks:
                    request._process_response_callbacks(response)

                has_listeners and notify(NewResponse(request, response))

                return response

            finally:
                if request.finished_callbacks:
                    request._process_finished_callbacks()

        finally:
            manager.pop()

 def test_it(self):
     from pyramid.threadlocal import manager
     try:
         manager.push({'registry':123})
         self.assertEqual(self._callFUT(), 123)
     finally:
         manager.pop()

    def __call__(self, environ, start_response):
        """
        Accept ``environ`` and ``start_response``; create a
        :term:`request` and route the request to a :app:`Pyramid`
        view based on introspection of :term:`view configuration`
        within the application registry; call ``start_response`` and
        return an iterable.
        """
        registry = self.registry
        manager = self.threadlocal_manager
        request = None
        threadlocals = {'registry':registry, 'request':request}
        manager.push(threadlocals)

        try:
            try:
                request = self.request_factory(environ)
                threadlocals['request'] = request
                request.registry = registry
                response = self.handle_request(request)
            finally:
                if request is not None and request.finished_callbacks:
                    request._process_finished_callbacks()

            return response(request.environ, start_response)
        finally:
            manager.pop()

    def run(self):
        if self.terminated:
            return

        manager.push({'registry': self.registry, 'request': None})
        loop = IOLoop.instance()
        ctx = zmq.Context()
        def callback():
            s = ctx.socket(zmq.PULL)
            s.setsockopt(zmq.LINGER, 0)
            s.bind(get_socket_url())
            def execute_next(action):
                # action is a list with one pickle
                method, obj = pickle.loads(action[0])
                # obj can be the DelayedCallback (dc)/Listener object or just
                # the identifier (event._p_oid) in case of the stop/close
                # method respectively.
                if method in ('stop', 'close', 'ack'):
                    # obj is actually here the identifier which was used to
                    # register the DelayedCallback/Listener
                    # (identifier attribute on the object)
                    identifier = obj
                    dc_or_listener = event_mod.callbacks.get(identifier, None)
                    if dc_or_listener is not None:
                        # stop DelayedCallback or close Listener
                        if method != 'ack':
                            getattr(dc_or_listener, method)()

                        del event_mod.callbacks[identifier]
                else:
                    # system crawler doesn't have an identifier
                    # and so the DelayedCallback started from a SignalEvent
                    if obj.identifier is not None:
                        event_mod.callbacks[obj.identifier] = obj

                    # mainly some_delayed_callback.start_in_ioloop()
                    getattr(obj, method)()

            self.stream = ZMQStream(s)
            self.stream.on_recv(execute_next)

        # It's ok to not use loop.add_callback
        # (the only method that is thread safe)
        # because the loop as not started yet
        loop.add_timeout(loop.time() + 2, callback)

        db = self.event.database
        root = db.open().root()['app_root']
        start_intermediate_events(root)
        root._p_jar.close()
        try:
            loop.start()
        except zmq.ZMQError:
            # only needed for tests isolation
            # we go here only if we try to close the stream in the stop method
            # below
            loop._callbacks = []
            loop._timeouts = []
            raise

def _test_load_fixture(request, discussion, admin, fixture):
    manager.push({'request': request})
    request.matchdict = {'discussion_slug': discussion.slug}
    json = fixture.generic_json(permissions=(P_SYSADMIN, ))
    print fixture.__dict__
    fixture.update_from_json(json, admin.id)
    print fixture.__dict__
    assert not discussion.db.is_modified(fixture, True)

 def setUp(self):
     from pyramid.threadlocal import manager
     from pyramid.registry import Registry
     manager.clear()
     registry = Registry('testing')
     self.registry = registry
     manager.push({'registry':registry, 'request':None})
     from zope.deprecation import __show__
     __show__.off()

def prepare(request=None, registry=None):
    """ This function pushes data onto the Pyramid threadlocal stack
    (request and registry), making those objects 'current'.  It
    returns a dictionary useful for bootstrapping a Pyramid
    application in a scripting environment.

    ``request`` is passed to the :app:`Pyramid` application root
    factory to compute the root. If ``request`` is None, a default
    will be constructed using the registry's :term:`Request Factory`
    via the :meth:`pyramid.interfaces.IRequestFactory.blank` method.

    If ``registry`` is not supplied, the last registry loaded from
    :attr:`pyramid.config.global_registries` will be used. If you
    have loaded more than one :app:`Pyramid` application in the
    current process, you may not want to use the last registry
    loaded, thus you can search the ``global_registries`` and supply
    the appropriate one based on your own criteria.

    The function returns a dictionary composed of ``root``,
    ``closer``, ``registry``, ``request`` and ``root_factory``.  The
    ``root`` returned is the application's root resource object.  The
    ``closer`` returned is a callable (accepting no arguments) that
    should be called when your scripting application is finished
    using the root.  ``registry`` is the registry object passed or
    the last registry loaded into
    :attr:`pyramid.config.global_registries` if no registry is passed.
    ``request`` is the request object passed or the constructed request
    if no request is passed.  ``root_factory`` is the root factory used
    to construct the root.
    """
    if registry is None:
        registry = getattr(request, 'registry', global_registries.last)
    if registry is None:
        raise ConfigurationError('No valid Pyramid applications could be '
                                 'found, make sure one has been created '
                                 'before trying to activate it.')
    if request is None:
        request = _make_request('/', registry)
    # NB: even though _make_request might have already set registry on
    # request, we reset it in case someone has passed in their own
    # request.
    request.registry = registry 
    threadlocals = {'registry':registry, 'request':request}
    threadlocal_manager.push(threadlocals)
    extensions = registry.queryUtility(IRequestExtensions)
    if extensions is not None:
        request._set_extensions(extensions)
    def closer():
        threadlocal_manager.pop()
    root_factory = registry.queryUtility(IRootFactory,
                                         default=DefaultRootFactory)
    root = root_factory(request)
    if getattr(request, 'context', None) is None:
        request.context = root
    return {'root':root, 'closer':closer, 'registry':registry,
            'request':request, 'root_factory':root_factory}

def execute_callback(app, callback, login):
    # set site and interaction that will be memorized in job
    request = DummyRequest()
    request.root = app
    registry = get_current_registry()
    manager.push({'registry': registry, 'request': request})
    user = get_user_by_login(login, request)
    request.user = user
    callback()
    manager.pop()

 def setUp(self):
     registry = get_current_registry()
     db = registry._zodb_databases[self.database_name]
     app = db.open().root()[self.site_id]
     request = DummyRequest()
     request.root = app
     manager.push({'registry': registry, 'request': request})
     user = get_user_by_userid(self.userid)
     request.user = user
     return app

 def test_add_translation_dirs_registers_chameleon_translate(self):
     from pyramid.interfaces import IChameleonTranslate
     from pyramid.threadlocal import manager
     request = DummyRequest()
     config = self._makeOne(autocommit=True)
     manager.push({'request':request, 'registry':config.registry})
     try:
         config.add_translation_dirs('pyramid.tests.pkgs.localeapp:locale')
         translate = config.registry.getUtility(IChameleonTranslate)
         self.assertEqual(translate('Approve'), u'Approve')
     finally:
         manager.pop()

    def test_lookup_with_request_pluralize(self):
        request = DummyRequest()
        localizer = get_localizer(request)

        info = defaults()
        info['request'] = request
        info['registry'].settings = {}
        manager.push(info)

        lookup = DummyLookup()
        path = self._get_template_path('minimal.genshi')
        renderer = self.make_one(path, lookup)
        self.assertEqual(renderer.adaptor.pluralize, localizer.pluralize)

 def test_it_defaults(self):
     from pyramid.threadlocal import manager
     from pyramid.threadlocal import get_current_registry
     from pyramid.registry import Registry
     old = True
     manager.push(old)
     config = self._callFUT()
     current = manager.get()
     self.assertFalse(current is old)
     self.assertEqual(config.registry, current['registry'])
     self.assertEqual(current['registry'].__class__, Registry)
     self.assertEqual(current['request'], None)
     self._assertSMHook(get_current_registry)

 def _threadContext(self):
   # IMPORTANT: this assumes that APScheduler invokes jobs in a separate
   #            thread per job (as documented)...
   # TODO: this needs confirmation!
   if self.appreg is None:
     yield
   else:
     from pyramid.threadlocal import manager
     reg = dict(manager.get())
     reg['registry'] = self.appreg
     manager.push(reg)
     try:
       yield
     finally:
       manager.pop()

 def test_registry_cannot_be_inited(self):
     from pyramid.threadlocal import manager
     registry = DummyRegistry()
     def raiseit(name):
         raise TypeError
     registry.__init__ = raiseit
     old = {'registry':registry}
     try:
         manager.push(old)
         self._callFUT() # doesn't blow up
         current = manager.get()
         self.assertNotEqual(current, old)
         self.assertEqual(registry.inited, 1)
     finally:
         manager.clear()

 def test_defaults(self):
     from pyramid.threadlocal import manager
     registry = DummyRegistry()
     old = {'registry':registry}
     hook = lambda *arg: None
     try:
         self._setSMHook(hook)
         manager.push(old)
         self._callFUT()
         current = manager.get()
         self.assertNotEqual(current, old)
         self.assertEqual(registry.inited, 2)
     finally:
         result = getSiteManager.sethook(None)
         self.assertNotEqual(result, hook)

    def run(self):
        # start dispatcher
        self.registry.__smxq_dispatcher__.start()

        # thread locals
        threadlocals = {'registry': self.registry,
                        'request': self.registry.__smxq_dispatcher__.request}
        threadlocal_manager.push(threadlocals)

        print 'Smxq dispatcher is initialized... workers are started...'

        try:
            while 1:
                gevent.sleep(1.0)
        except KeyboardInterrupt:
            pass

 def test_it_raises_if_no_registry(self):
     request = self._makeOne()
     del request.registry
     from pyramid.threadlocal import manager
     manager.push({'registry': None, 'request': request})
     try:
         raise RuntimeError
     except RuntimeError:
         try:
             request.invoke_exception_view()
         except RuntimeError as e:
             self.assertEqual(e.args[0], "Unable to retrieve registry")
     else: # pragma: no cover
         self.fail()
     finally:
         manager.pop()

    def invoke_subrequest(self, request, use_tweens=False):
        """Obtain a response object from the Pyramid application based on
        information in the ``request`` object provided.  The ``request``
        object must be an object that implements the Pyramid request
        interface (such as a :class:`pyramid.request.Request` instance).  If
        ``use_tweens`` is ``True``, the request will be sent to the
        :term:`tween` in the tween stack closest to the request ingress.  If
        ``use_tweens`` is ``False``, the request will be sent to the main
        router handler, and no tweens will be invoked.
        
        See the API for pyramid.request for complete documentation.
        """
        registry = self.registry
        has_listeners = self.registry.has_listeners
        notify = self.registry.notify
        threadlocals = {'registry':registry, 'request':request}
        manager = self.threadlocal_manager
        manager.push(threadlocals)
        request.registry = registry
        request.invoke_subrequest = self.invoke_subrequest
        
        if use_tweens:
            handle_request = self.handle_request
        else:
            handle_request = self.orig_handle_request

        try:

            try:
                extensions = self.request_extensions
                if extensions is not None:
                    apply_request_extensions(request, extensions=extensions)
                response = handle_request(request)

                if request.response_callbacks:
                    request._process_response_callbacks(response)

                has_listeners and notify(NewResponse(request, response))
                
                return response

            finally:
                if request.finished_callbacks:
                    request._process_finished_callbacks()

        finally:
            manager.pop()