[python-ldap] SyncreplConsumer running in parallel with other application

[David Gabriel]

I am under ubuntu, python2.7, I have installed python-ldap few mounths ago
..
Here is my code:

#!/usr/bin/python

# Import the python-ldap modules
import ldap,ldapurl
# Import specific classes from python-ldap
from ldap.ldapobject import ReconnectLDAPObject
from ldap.syncrepl import SyncreplConsumer

# Import modules from Python standard lib
import shelve,signal,time,sys,logging
import eventlet
import thread

# Global state
watcher_running = True
ldap_connection = False



class SyncReplConsumer(ReconnectLDAPObject,SyncreplConsumer):
    """
    Syncrepl Consumer interface
    """
    def __init__(self,db_path,*args,**kwargs):
        # Initialise the LDAP Connection first
        ldap.ldapobject.ReconnectLDAPObject.__init__(self, *args, **kwargs)
        # Now prepare the data store
        self.__data = shelve.open(db_path, 'c')
        # We need this for later internal use
        self.__presentUUIDs = dict()

    def __del__(self):
            # Close the data store properly to avoid corruption
            self.__data.close()

    def syncrepl_get_cookie(self):
        if 'cookie' in self.__data:
            return self.__data['cookie']

    def syncrepl_set_cookie(self,cookie):
        self.__data['cookie'] = cookie

    def syncrepl_entry(self,dn,attributes,uuid):
        global result
        # First we determine the type of change we have here (and store
away the previous data for later if needed)
        previous_attributes = dict()
        if uuid in self.__data:
            change_type = 'modify'
            previous_attributes = self.__data[uuid]
        else:
            change_type = 'add'
        # Now we store our knowledge of the existence of this entry
(including the DN as an attribute for convenience)
        attributes['dn'] = dn
        self.__data[uuid] = attributes
        # Debugging
        print 'Detected', change_type, 'of entry:', dn
        # If we have a cookie then this is not our first time being run, so
it must be a change
        if 'ldap_cookie' in self.__data:
                self.perform_application_sync(dn, attributes,
previous_attributes)

    def syncrepl_delete(self,uuids):
        global result
        # Make sure we know about the UUID being deleted, just in case...
        uuids = [uuid for uuid in uuids if uuid in self.__data]
        # Delete all the UUID values we know of
        for uuid in uuids:
            print 'Detected deletion of entry:', self.__data[uuid]['dn']
            del self.__data[uuid]

    def syncrepl_present(self,uuids,refreshDeletes=False):
        # If we have not been given any UUID values, then we have recieved
all the present controls...
        if uuids is None:
            # We only do things if refreshDeletes is false as the syncrepl
extension will call syncrepl_delete instead when it detects a delete notice
            if refreshDeletes is False:
                deletedEntries = [uuid for uuid in self.__data.keys() if
uuid not in self.__presentUUIDs and uuid != 'ldap_cookie']
                self.syncrepl_delete( deletedEntries )
            # Phase is now completed, reset the list
            self.__presentUUIDs = {}
        else:
            # Note down all the UUIDs we have been sent
            for uuid in uuids:
                    self.__presentUUIDs[uuid] = True

    def perform_application_sync(self,dn,attributes,previous_attributes):
        print 'Performing application sync for:', dn
        return True


# Shutdown handler
#def commenceShutdown(signum, stack):
def commenceShutdown():
    # Declare the needed global variables
    global watcher_running, ldap_connection
    print 'Shutting down!'

    # We are no longer running
    watcher_running = False

    # Tear down the server connection
    if( ldap_connection ):
            del ldap_connection

    # Shutdown
    sys.exit(0)

def mainOfSyncrepl(threadName):
    # Time to actually begin execution
    # Install our signal handlers
#    signal.signal(signal.SIGTERM,commenceShutdown)
#    signal.signal(signal.SIGINT,commenceShutdown)
    try:
      ldap_url =
ldapurl.LDAPUrl('ldap://Address/dc=test,dc=com?*?sub?(objectClass=*)?bindname=cn=admin%2cdc=test%2cdc=com,X-BINDPW=password')#ldapurl.LDAPUrl(sys.argv[1])
    #  ldap_url = ldapurl.LDAPUrl(link)
      database_path = 'example.com'#sys.argv[2]
    #  database_path = pathName
    except IndexError,e:
      print 'Usage: syncrepl-client.py <LDAP URL> <pathname of database>'
      sys.exit(1)
    except ValueError,e:
      print 'Error parsing command-line arguments:',str(e)
      sys.exit(1)

    while watcher_running:
        print 'Connecting to LDAP server now...'
        # Prepare the LDAP server connection (triggers the connection as
well)
        ldap_connection =
SyncReplConsumer(database_path,ldap_url.initializeUrl())

        # Now we login to the LDAP server
        try:
        ldap_connection.simple_bind_s(ldap_url.who,ldap_url.cred)
        except ldap.INVALID_CREDENTIALS, e:
        print 'Login to LDAP server failed: ', str(e)
        sys.exit(1)
        except ldap.SERVER_DOWN:
        continue

        # Commence the syncing
        print 'Commencing sync process'
        ldap_search = ldap_connection.syncrepl_search(
          ldap_url.dn or '',
          ldap_url.scope or ldap.SCOPE_SUBTREE,
          mode = 'refreshAndPersist',
          filterstr = ldap_url.filterstr or '(objectClass=*)'
        )
        print 'After syncrepl_search.'
        try:
        while ldap_connection.syncrepl_poll( all = 1, msgid = ldap_search):
            pass
        except KeyboardInterrupt:
        # User asked to exit
        commenceShutdown()
        pass
        except Exception, e:
        # Handle any exception
        if watcher_running:
            print 'Encountered a problem, going to retry. Error:', str(e)
            time.sleep(5)
        pass

# Define a function for the 2nd thread
def print_time(ThreadName):
        count = 0
    delay = 3
        while 1:#count < 5:
            count += 1
            print "%s: %s" % (ThreadName, time.ctime(time.time()) )
            time.sleep(delay)



print 'Before call threads'

syncreplEvt1 = eventlet.spawn(mainOfSyncrepl, "Thread-1",)
syncreplEvt2 = eventlet.spawn(print_time, "Thread-2",)
#syncreplEvt3 = eventlet.spawn(print_time, "Thread-3",)

print 'After call threads'

syncreplEvt1.wait()
syncreplEvt2.wait()
#syncreplEvt3.wait()

2016-01-08 7:56 GMT-08:00 Michael Ströder <michael at stroeder.com>:

> David Gabriel wrote:
> > After some investigation in the source code I think the function
> 'result4'
> > located in the file ldapobject.py is the responsible of this blocking
> > behaviour.
>
> 1. Please note that I personally do not feel responsible for the code
> *fork* on
> github.
>
> 2. You did not tell us any other relevant details like
> - OS
> - Python version
> - python-ldap version (from where?)
> - whether python-ldap is linked against libldap_r etc.
>
> While method LDAPObject.result4() is called in a blocking manner it's IMHO
> likely not the cause for your code not being executed in parallel.
>
> Frankly I don't see any reason why it's not possible to run the blocking
> syncrepl client code in one thread and having other code running in
> separate
> threads.
>
> It might help if you would post a *small* code example with threads
> illustrating
> what you're trying to achieve and which shows your blocking issue.
>
> Ciao, Michael.
>
>
-------------- next part --------------
An HTML attachment was scrubbed...
URL: <http://mail.python.org/pipermail/python-ldap/attachments/20160108/e235d5f7/attachment.html>