Commit ee110eaa authored by Sebastien Robin's avatar Sebastien Robin

Commit the work of Fabien

- use list of string instead of adding string to strings
  in order to improve performance
- use message id instead of session id. Increase message id
  for each message
- start to implement authentication


git-svn-id: https://svn.erp5.org/repos/public/erp5/trunk@14280 20353a03-c40f-0410-a6d1-a30d3c3de9de
parent b3833444
......@@ -39,12 +39,21 @@ from zLOG import LOG
class PublicationSynchronization(XMLSyncUtils):
def PubSyncInit(self, publication=None, xml_client=None, subscriber=None, sync_type=None):
def PubSyncInit(self, publication=None, xml_client=None, subscriber=None,
sync_type=None, auth_required=0):
"""
Read the client xml message
Send the first XML message from the server
"""
LOG('PubSyncInit',0,'Starting... publication: %s' % str(publication))
#the session id is set at the same value of those of the client
subscriber.setSessionId(self.getSessionId(xml_client))
# for a new session, the message Id must be reset
subscriber.resetMessageId()
#the last_message_id is 1 because the message that
#we are about to send is the message 1
subscriber.initLastMessageId(1)
alert = None
# Get informations from the body
......@@ -53,55 +62,91 @@ class PublicationSynchronization(XMLSyncUtils):
next_anchor = self.getAlertNextAnchor(xml_client)
alert = self.checkAlert(xml_client)
alert_code = self.getAlertCode(xml_client)
# If slow sync, then resend everything
if alert_code == self.SLOW_SYNC:
LOG('Warning !!!, reseting client synchronization for subscriber:',0,subscriber)
subscriber.resetAllSignatures()
# Check if the last time synchronization is the same as the client one
if subscriber.getNextAnchor() != last_anchor:
if last_anchor == None:
LOG('PubSyncInit',0,'anchor null')
cred = self.checkCred(xml_client)
#XXX this is in developement, it's just for tests
if not cred and auth_required:
LOG('PubSyncInit',0,'authentication required')
# Prepare the xml message for the Sync initialization package
cmd_id = 1 # specifies a SyncML message-unique command identifier
xml_list = []
xml = xml_list.append
xml('<SyncML>\n')
# syncml header
xml(self.SyncMLHeader(subscriber.getSessionId(),
subscriber.incrementMessageId(), subscriber.getSubscriptionUrl(),
publication.getPublicationUrl()))
# syncml body
xml(' <SyncBody>\n')
# chal message
xml(self.SyncMLChal(cmd_id, "SyncHdr", publication.getPublicationUrl(),
subscriber.getSubscriptionUrl(), "b64", "syncml:auth-basic",
self.UNAUTHORIZED))
cmd_id += 1
xml(' </SyncBody>\n')
xml('</SyncML>\n')
xml_a = ''.join(xml_list)
self.sendResponse(from_url=publication.getPublicationUrl(),
to_url=subscriber.getSubscriptionUrl(),sync_id=publication.getTitle(),
xml=xml_a,domain=publication)
else :
# If slow sync, then resend everything
if alert_code == self.SLOW_SYNC:
LOG('Warning !!!, reseting client synchronization for subscriber:',0,
subscriber)
subscriber.resetAllSignatures()
# Check if the last time synchronization is the same as the client one
mess='\nsubscriber.getNextAnchor:\t%s\nsubscriber.getLastAnchor:\t%s\
\nlast_anchor:\t\t\t%s\nnext_anchor:\t\t\t%s' % (subscriber.getNextAnchor(),
subscriber.getLastAnchor(), last_anchor, next_anchor)
LOG('PubSyncInit',0,mess)
if subscriber.getNextAnchor() != last_anchor:
if last_anchor == None:
LOG('PubSyncInit',0,'anchor null')
raise ValueError, "Sorry, the anchor was null"
else:
message = "bad anchors in PubSyncInit! " + subscriber.getNextAnchor() + \
" and " + last_anchor
LOG('PubSyncInit',0,message)
else:
message = "bad anchors in PubSyncInit! " + subscriber.getNextAnchor() + \
" and " + last_anchor
LOG('PubSyncInit',0,message)
else:
subscriber.setNextAnchor(next_anchor)
subscriber.setNextAnchor(next_anchor)
# We have to set every object as NOT_SYNCHRONIZED
subscriber.startSynchronization()
else:
# We have started the sync from the server (may be for a conflict resolution)
pass
xml = ""
#if alert is not None:
if 1:
if alert is not None and auth_required==0:
#if 1:
# Prepare the xml message for the Sync initialization package
cmd_id = 1 # specifies a SyncML message-unique command identifier
xml = ""
xml += '<SyncML>\n'
xml_list = []
xml = xml_list.append
xml('<SyncML>\n')
# syncml header
xml += self.SyncMLHeader(subscriber.getSessionId(), "1",
subscriber.getSubscriptionUrl(), publication.getPublicationUrl())
xml(self.SyncMLHeader(subscriber.getSessionId(),
subscriber.incrementMessageId(), subscriber.getSubscriptionUrl(),
publication.getPublicationUrl()))
# syncml body
xml += ' <SyncBody>\n'
xml(' <SyncBody>\n')
# alert message
xml += self.SyncMLAlert(cmd_id, sync_type, subscriber.getSubscriptionUrl(),
publication.getPublicationUrl(), subscriber.getLastAnchor(), subscriber.getNextAnchor())
xml(self.SyncMLAlert(cmd_id, sync_type, subscriber.getSubscriptionUrl(),
publication.getPublicationUrl(), subscriber.getLastAnchor(),
subscriber.getNextAnchor()))
cmd_id += 1
xml += ' </SyncBody>\n'
xml += '</SyncML>\n'
xml(' </SyncBody>\n')
xml('</SyncML>\n')
xml_a = ''.join(xml_list)
self.sendResponse(from_url=publication.getPublicationUrl(),
to_url=subscriber.getSubscriptionUrl(),sync_id=publication.getTitle(),xml=xml,
domain=publication)
return {'has_response':1,'xml':xml}
self.sendResponse(from_url=publication.getPublicationUrl(),
to_url=subscriber.getSubscriptionUrl(), sync_id=publication.getTitle(),
xml=xml_a, domain=publication)
return {'has_response':1,'xml':xml_a}
def PubSync(self, id, msg=None, RESPONSE=None, subscriber=None):
......@@ -118,23 +163,27 @@ class PublicationSynchronization(XMLSyncUtils):
publication = self.getPublication(id)
if xml_client is not None:
if type(xml_client) in (type('a'),type(u'a')):
if isinstance(xml_client, str) or isinstance(xml_client, unicode):
xml_client = parseString(xml_client)
first_node = xml_client.childNodes[0]
if first_node.nodeName != "SyncML":
LOG('PubSync',0,'This is not a SyncML Message')
return
raise ValueError, "Sorry, This is not a SyncML Message"
alert_code = self.getAlertCode(xml_client)
# Get informations from the header
client_header = first_node.childNodes[1]
if client_header.nodeName != "SyncHdr":
LOG('PubSync',0,'This is not a SyncML Header')
return
raise ValueError, "Sorry, This is not a SyncML Header"
for subnode in client_header.childNodes:
if subnode.nodeType == subnode.ELEMENT_NODE and subnode.nodeName == "Source":
subscription_url = str(subnode.childNodes[0].data)
if subnode.nodeType == subnode.ELEMENT_NODE and \
subnode.nodeName == "Source":
for subnode2 in subnode.childNodes:
if subnode2.nodeType == subnode2.ELEMENT_NODE and \
subnode2.nodeName == "LocURI":
subscription_url = str(subnode2.childNodes[0].data)
# Get the subscriber or create it if not already in the list
subscriber = publication.getSubscriber(subscription_url)
if subscriber == None:
......@@ -142,19 +191,21 @@ class PublicationSynchronization(XMLSyncUtils):
subscriber.setXMLMapping(publication.getXMLMapping())
publication.addSubscriber(subscriber)
# first synchronization
result = self.PubSyncInit(publication,xml_client,subscriber=subscriber,sync_type=self.SLOW_SYNC)
result = self.PubSyncInit(publication,xml_client,subscriber=subscriber,
sync_type=self.SLOW_SYNC)
elif self.checkAlert(xml_client) and alert_code in (self.TWO_WAY,self.SLOW_SYNC):
result = self.PubSyncInit(publication=publication,
xml_client=xml_client, subscriber=subscriber,sync_type=alert_code)
elif self.checkAlert(xml_client) and \
alert_code in (self.TWO_WAY,self.SLOW_SYNC):
result = self.PubSyncInit(publication=publication,
xml_client=xml_client, subscriber=subscriber, sync_type=alert_code)
else:
result = self.PubSyncModif(publication, xml_client)
elif subscriber is not None:
# This looks like we are starting a synchronization after
# a conflict resolution by the user
result = self.PubSyncInit(publication=publication,
xml_client=None, subscriber=subscriber,sync_type=self.TWO_WAY)
result = self.PubSyncInit(publication=publication, xml_client=None,
subscriber=subscriber,sync_type=self.TWO_WAY)
if RESPONSE is not None:
RESPONSE.redirect('managePublications')
......
......@@ -660,7 +660,7 @@ class Subscription(Folder, SyncCode):
self.setConduit(conduit)
Folder.__init__(self, id)
self.title = title
#self.signatures = PersitentMapping()
def getTitle(self):
......@@ -723,6 +723,26 @@ class Subscription(Folder, SyncCode):
self.last_session_id = session_id
return 1
def checkCorrectRemoteMessageId(self, message_id):
"""
We will see if the last message id was the same
wich means that the same message was sent again
return 1 if the message id was not seen, 0 if already seen
"""
last_message_id = getattr(self,'last_message_id',None)
LOG('checkCorrectRemoteMessageId last_message_id =',0,last_message_id)
LOG('checkCorrectRemoteMessageId message_id =',0,message_id)
if last_message_id == message_id:
return 0
self.last_message_id = message_id
return 1
def initLastMessageId(self, last_message_id=None):
"""
set the last message id to 0
"""
self.last_message_id=last_message_id
def getLastSentMessage(self):
"""
......@@ -926,7 +946,7 @@ class Subscription(Folder, SyncCode):
query_list = []
if query is None:
return query_list
if type(query) is type('a'):
if isinstance(query, str):
query_method = getattr(destination,query,None)
if query_method is not None:
query_list = query_method()
......@@ -1009,14 +1029,51 @@ class Subscription(Folder, SyncCode):
return the current subscription
"""
return self
def setSessionId(self, session_id):
"""
set the session id
"""
self.session_id = session_id
def getSessionId(self):
"""
return the session id
"""
#self.session_id += 1 #to be commented
return self.session_id
def incrementSessionId(self):
"""
increment and return the session id
"""
self.session_id += 1
self.resetMessageId() # for a new session, the message Id must be reset
return self.session_id
def incrementMessageId(self):
"""
return the message id
"""
#self.message_id += 1
#return self.message_id
#return 5
value = getattr(self, 'message_id', 0)
self.message_id = value +1
return self.message_id
def getMessageId(self):
"""
increment and return the message id
"""
return self.message_id
def resetMessageId(self):
"""
set the message id to 0
"""
self.message_id = 0
def getLastAnchor(self):
"""
return the id of the last synchronisation
......
......@@ -44,38 +44,43 @@ class SubscriptionSynchronization(XMLSyncUtils):
"""
LOG('SubSyncInit',0,'starting....')
cmd_id = 1 # specifies a SyncML message-unique command identifier
xml = ""
xml += '<SyncML>\n'
xml_list = []
xml = xml_list.append
xml('<SyncML>\n')
# syncml header
xml += self.SyncMLHeader(subscription.getSessionId(), "1",
subscription.getPublicationUrl(), subscription.getSubscriptionUrl())
xml(self.SyncMLHeader(subscription.incrementSessionId(),
subscription.incrementMessageId(), subscription.getPublicationUrl(),
subscription.getSubscriptionUrl()))
# syncml body
xml += ' <SyncBody>\n'
xml(' <SyncBody>\n')
subscription.NewAnchor()
subscription.initLastMessageId()
# We have to set every object as NOT_SYNCHRONIZED
subscription.startSynchronization()
# alert message
xml += self.SyncMLAlert(cmd_id, subscription.getSynchronizationType(),
xml(self.SyncMLAlert(cmd_id, subscription.getSynchronizationType(),
subscription.getPublicationUrl(),
subscription.getDestinationPath(),
subscription.getLastAnchor(), subscription.getNextAnchor())
subscription.getLastAnchor(),
subscription.getNextAnchor()))
cmd_id += 1
xml += ' <Put>\n'
xml += ' <CmdID>%s</CmdID>\n' % cmd_id ; cmd_id += 1
xml += ' </Put>\n'
xml += ' </SyncBody>\n'
xml += '</SyncML>\n'
xml(' <Put>\n')
xml(' <CmdID>%s</CmdID>\n' % cmd_id)
cmd_id += 1
xml(' </Put>\n')
xml(' </SyncBody>\n')
xml('</SyncML>\n')
xml_a = ''.join(xml_list)
self.sendResponse(from_url=subscription.subscription_url, to_url=subscription.publication_url,
sync_id=subscription.getTitle(), xml=xml,domain=subscription)
self.sendResponse(from_url=subscription.subscription_url,
to_url=subscription.publication_url, sync_id=subscription.getTitle(),
xml=xml_a,domain=subscription)
return {'has_response':1,'xml':xml}
return {'has_response':1,'xml':xml_a}
def SubSync(self, id, msg=None, RESPONSE=None):
"""
......@@ -83,17 +88,17 @@ class SubscriptionSynchronization(XMLSyncUtils):
"""
LOG('SubSync',0,'starting... id: %s' % str(id))
LOG('SubSync',0,'starting... msg: %s' % str(msg))
response = None #check if subsync replies to this messages
subscription = self.getSubscription(id)
if msg==None and (subscription.getSubscriptionUrl()).find('file')>=0:
msg = self.readResponse(sync_id=id,from_url=subscription.getSubscriptionUrl())
msg = self.readResponse(sync_id=id,
from_url=subscription.getSubscriptionUrl())
if msg==None:
response = self.SubSyncInit(self.getSubscription(id))
else:
xml_client = msg
if type(xml_client) in (type('a'),type(u'a')):
if isinstance(xml_client, str) or isinstance(xml_client, unicode):
xml_client = parseString(xml_client)
response = self.SubSyncModif(self.getSubscription(id),xml_client)
......
......@@ -47,6 +47,9 @@ class SyncCode(Persistent):
# whatever is needed to change(replace)
CONFLICT_CLIENT_WIN = 208 # The client is the "winner", we keep
# the version of the client
UNAUTHORIZED = 401
AUTH_REQUIRED = 407
AUTH_ACCEPTED = 212
# Difference between publication and subscription
PUB = 1
......
......@@ -734,9 +734,9 @@ class SynchronizationTool( SubscriptionSynchronization, PublicationSynchronizati
"""
if context is None:
return context
elif type(context) is type(()):
elif isinstance(context, tuple):
return context
elif type(context) is type('a'):
elif isinstance(context, tuple):
return tuple(context.split('/'))
else:
return context.getPhysicalPath()
......@@ -751,7 +751,7 @@ class SynchronizationTool( SubscriptionSynchronization, PublicationSynchronizati
LOG('sendResponse, to_url: ',0,to_url)
LOG('sendResponse, from_url: ',0,from_url)
LOG('sendResponse, sync_id: ',0,sync_id)
LOG('sendResponse, xml: ',0,xml)
LOG('sendResponse, xml: \n',0,xml)
if domain is not None:
gpg_key = domain.getGPGKey()
if gpg_key not in ('',None):
......@@ -768,14 +768,14 @@ class SynchronizationTool( SubscriptionSynchronization, PublicationSynchronizati
commands.getstatusoutput('rm -f /tmp/%s.gz' % filename)
commands.getstatusoutput('rm -f /tmp/%s.gz.gpg' % filename)
if send:
if type(to_url) is type('a'):
if isinstance(to_url, str):
if to_url.find('http://')==0:
# XXX Make sure this is not a problem
if domain.domain_type == self.PUB:
return None
# we will send an http response
domain = aq_base(domain)
LOG('sendResponse, will start sendHttpResponse, xml',0,xml)
LOG('sendResponse, will start sendHttpResponse, xml\n',0,xml)
self.activate(activity='RAMQueue').sendHttpResponse(sync_id=sync_id,
to_url=to_url,
xml=xml, domain=domain)
......@@ -907,30 +907,38 @@ class SynchronizationTool( SubscriptionSynchronization, PublicationSynchronizati
commands.getstatusoutput('rm -f /tmp/%s.gz.gpg' % filename)
# Get the target and then find the corresponding publication or
# Subscription
LOG('readResponse, xml before parseSTring',0,text)
LOG('readResponse, xml before parseSTring\n',0,text)
xml = parseString(text)
url = ''
for subnode in self.getElementNodeList(xml):
if subnode.nodeName == 'SyncML':
for subnode1 in self.getElementNodeList(subnode):
if subnode1.nodeName == 'SyncHdr':
for subnode2 in self.getElementNodeList(subnode1):
if subnode2.nodeName == 'Target':
url = subnode2.childNodes[0].data
#XXX this function is not very optimized and should be improved
url = self.getTarget(xml)
for publication in self.getPublicationList():
if publication.getPublicationUrl()==url and publication.getTitle()==sync_id:
result = self.PubSync(sync_id,xml)
# Then encrypt the message
xml = result['xml']
xml = self.sendResponse(xml=xml,domain=publication,send=0)
#must be commented because this method is alredy called
#xml = self.sendResponse(xml=xml,domain=publication,send=0)
return xml
for subscription in self.getSubscriptionList():
if subscription.getSubscriptionUrl()==url and subscription.getTitle()==sync_id:
result = self.activate(activity='RAMQueue').SubSync(sync_id,xml)
if subscription.getSubscriptionUrl()==url and \
subscription.getTitle()==sync_id:
next_status = self.getNextSyncBodyStatus(xml, None)
if next_status is not None:
status_code = self.getStatusCode(next_status)
LOG('readResponse status code :',0,status_code)
if status_code == self.UNAUTHORIZED or \
status_code == self.AUTH_REQUIRED:
LOG('readResponse', 0, 'Authentication required')
raise ValueError, "Authentication required"
else:
result = self.activate(activity='RAMQueue').SubSync(sync_id,xml)
#result = self.SubSync(sync_id,xml)
# we use from only if we have a file
elif type(from_url) is type('a'):
elif isinstance(from_url, str):
if from_url.find('file://')==0:
try:
filename = from_url[len('file:/'):]
......
This diff is collapsed.
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment