Refactored message status-related code
This commit is contained in:
Jonathan Warren
parent
c762d7ed01
commit
a0e6430c83
|
|
@ -21,7 +21,6 @@ import sys
|
|||
import ConfigParser
|
||||
import Queue
|
||||
from addresses import *
|
||||
#from shared import *
|
||||
import shared
|
||||
from defaultKnownNodes import *
|
||||
import time
|
||||
|
|
@ -321,19 +320,23 @@ class receiveDataThread(threading.Thread):
|
|||
#print 'self.data is currently ', repr(self.data)
|
||||
#shared.printLock.release()
|
||||
if len(self.data) < 20: #if so little of the data has arrived that we can't even unpack the payload length
|
||||
pass
|
||||
elif self.data[0:4] != '\xe9\xbe\xb4\xd9':
|
||||
return
|
||||
if self.data[0:4] != '\xe9\xbe\xb4\xd9':
|
||||
if verbose >= 1:
|
||||
shared.printLock.acquire()
|
||||
sys.stderr.write('The magic bytes were not correct. First 40 bytes of data: %s\n' % repr(self.data[0:40]))
|
||||
print 'self.data:', self.data.encode('hex')
|
||||
shared.printLock.release()
|
||||
self.data = ""
|
||||
else:
|
||||
return
|
||||
self.payloadLength, = unpack('>L',self.data[16:20])
|
||||
if len(self.data) >= self.payloadLength+24: #check if the whole message has arrived yet. If it has,...
|
||||
if self.data[20:24] == hashlib.sha512(self.data[24:self.payloadLength+24]).digest()[0:4]:#test the checksum in the message. If it is correct...
|
||||
#print 'message checksum is correct'
|
||||
if len(self.data) < self.payloadLength+24: #check if the whole message has arrived yet.
|
||||
return
|
||||
if self.data[20:24] != hashlib.sha512(self.data[24:self.payloadLength+24]).digest()[0:4]:#test the checksum in the message. If it is correct...
|
||||
print 'Checksum incorrect. Clearing this message.'
|
||||
self.data = self.data[self.payloadLength+24:]
|
||||
self.processData()
|
||||
return
|
||||
#The time we've last seen this node is obviously right now since we just received valid data from it. So update the knownNodes list so that other peers can be made aware of its existance.
|
||||
if self.initiatedConnection and self.connectionIsOrWasFullyEstablished: #The remote port is only something we should share with others if it is the remote node's incoming port (rather than some random operating-system-assigned outgoing port).
|
||||
shared.knownNodesLock.acquire()
|
||||
|
|
@ -413,9 +416,8 @@ class receiveDataThread(threading.Thread):
|
|||
if len(self.ackDataThatWeHaveYetToSend) > 0:
|
||||
self.data = self.ackDataThatWeHaveYetToSend.pop()
|
||||
self.processData()
|
||||
else:
|
||||
print 'Checksum incorrect. Clearing this message.'
|
||||
self.data = self.data[self.payloadLength+24:]
|
||||
|
||||
|
||||
|
||||
def isProofOfWorkSufficient(self,data,nonceTrialsPerByte=0,payloadLengthExtraBytes=0):
|
||||
if nonceTrialsPerByte < shared.networkDefaultProofOfWorkNonceTrialsPerByte:
|
||||
|
|
@ -941,8 +943,7 @@ class receiveDataThread(threading.Thread):
|
|||
shared.sqlReturnQueue.get()
|
||||
shared.sqlSubmitQueue.put('commit')
|
||||
shared.sqlLock.release()
|
||||
#self.emit(SIGNAL("updateSentItemStatusByAckdata(PyQt_PyObject,PyQt_PyObject)"),encryptedData[readPosition:],'Acknowledgement of the message received just now.')
|
||||
shared.UISignalQueue.put(('updateSentItemStatusByAckdata',(encryptedData[readPosition:],'Acknowledgement of the message received just now.')))
|
||||
shared.UISignalQueue.put(('updateSentItemStatusByAckdata',(encryptedData[readPosition:],'Acknowledgement of the message received just now. '+ unicode(strftime(shared.config.get('bitmessagesettings', 'timeformat'),localtime(int(time.time()))),'utf-8'))))
|
||||
return
|
||||
else:
|
||||
shared.printLock.acquire()
|
||||
|
|
@ -1152,7 +1153,7 @@ class receiveDataThread(threading.Thread):
|
|||
|
||||
#self.emit(SIGNAL("displayNewSentMessage(PyQt_PyObject,PyQt_PyObject,PyQt_PyObject,PyQt_PyObject,PyQt_PyObject,PyQt_PyObject)"),toAddress,'[Broadcast subscribers]',fromAddress,subject,message,ackdata)
|
||||
shared.UISignalQueue.put(('displayNewSentMessage',(toAddress,'[Broadcast subscribers]',fromAddress,subject,message,ackdata)))
|
||||
shared.workerQueue.put(('sendbroadcast',(fromAddress,subject,message)))
|
||||
shared.workerQueue.put(('sendbroadcast',''))
|
||||
|
||||
if self.isAckDataValid(ackData):
|
||||
print 'ackData is valid. Will process it.'
|
||||
|
|
@ -1505,11 +1506,17 @@ class receiveDataThread(threading.Thread):
|
|||
#We have received an inv message
|
||||
def recinv(self,data):
|
||||
totalNumberOfObjectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave = 0
|
||||
print 'number of keys(hosts) in numberOfObjectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHavePerPeer:', len(numberOfObjectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHavePerPeer)
|
||||
for key, value in numberOfObjectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHavePerPeer.items():
|
||||
totalNumberOfObjectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave += value
|
||||
if len(numberOfObjectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHavePerPeer) > 0:
|
||||
shared.printLock.acquire()
|
||||
print 'number of keys(hosts) in numberOfObjectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHavePerPeer:', len(numberOfObjectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHavePerPeer)
|
||||
print 'totalNumberOfObjectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave = ', totalNumberOfObjectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave
|
||||
shared.printLock.release()
|
||||
numberOfItemsInInv, lengthOfVarint = decodeVarint(data[:10])
|
||||
if numberOfItemsInInv > 50000:
|
||||
sys.stderr.write('Too many items in inv message!')
|
||||
return
|
||||
if len(data) < lengthOfVarint + (numberOfItemsInInv * 32):
|
||||
print 'inv message doesn\'t contain enough data. Ignoring.'
|
||||
return
|
||||
|
|
@ -2562,7 +2569,7 @@ class singleCleaner(threading.Thread):
|
|||
shared.sqlSubmitQueue.put('commit')
|
||||
|
||||
t = ()
|
||||
shared.sqlSubmitQueue.put('''select toaddress, toripe, fromaddress, subject, message, ackdata, lastactiontime, status, pubkeyretrynumber, msgretrynumber FROM sent WHERE ((status='findingpubkey' OR status='sentmessage') AND folder='sent') ''') #If the message's folder='trash' then we'll ignore it.
|
||||
shared.sqlSubmitQueue.put('''select toaddress, toripe, fromaddress, subject, message, ackdata, lastactiontime, status, pubkeyretrynumber, msgretrynumber FROM sent WHERE ((status='awaitingpubkey' OR status='msgsent') AND folder='sent') ''') #If the message's folder='trash' then we'll ignore it.
|
||||
shared.sqlSubmitQueue.put(t)
|
||||
queryreturn = shared.sqlReturnQueue.get()
|
||||
for row in queryreturn:
|
||||
|
|
@ -2573,28 +2580,31 @@ class singleCleaner(threading.Thread):
|
|||
shared.printLock.release()
|
||||
break
|
||||
toaddress, toripe, fromaddress, subject, message, ackdata, lastactiontime, status, pubkeyretrynumber, msgretrynumber = row
|
||||
if status == 'findingpubkey':
|
||||
if status == 'awaitingpubkey':
|
||||
if int(time.time()) - lastactiontime > (maximumAgeOfAnObjectThatIAmWillingToAccept * (2 ** (pubkeyretrynumber))):
|
||||
print 'It has been a long time and we haven\'t heard a response to our getpubkey request. Sending again.'
|
||||
try:
|
||||
del neededPubkeys[toripe] #We need to take this entry out of the neededPubkeys structure because the shared.workerQueue checks to see whether the entry is already present and will not do the POW and send the message because it assumes that it has already done it recently.
|
||||
except:
|
||||
pass
|
||||
shared.workerQueue.put(('sendmessage',toaddress))
|
||||
|
||||
#self.emit(SIGNAL("updateStatusBar(PyQt_PyObject)"),"Doing work necessary to again attempt to request a public key...")
|
||||
shared.UISignalQueue.put(('updateStatusBar','Doing work necessary to again attempt to request a public key...'))
|
||||
t = (int(time.time()),pubkeyretrynumber+1,toripe)
|
||||
shared.sqlSubmitQueue.put('''UPDATE sent SET lastactiontime=?, pubkeyretrynumber=? WHERE toripe=?''')
|
||||
shared.sqlSubmitQueue.put('''UPDATE sent SET lastactiontime=?, pubkeyretrynumber=?, status='msgqueued' WHERE toripe=?''')
|
||||
shared.sqlSubmitQueue.put(t)
|
||||
shared.sqlReturnQueue.get()
|
||||
else:# status == sentmessage
|
||||
shared.sqlSubmitQueue.put('commit')
|
||||
shared.workerQueue.put(('sendmessage',''))
|
||||
else:# status == msgsent
|
||||
if int(time.time()) - lastactiontime > (maximumAgeOfAnObjectThatIAmWillingToAccept * (2 ** (msgretrynumber))):
|
||||
print 'It has been a long time and we haven\'t heard an acknowledgement to our msg. Sending again.'
|
||||
t = (int(time.time()),msgretrynumber+1,'findingpubkey',ackdata)
|
||||
t = (int(time.time()),msgretrynumber+1,'msgqueued',ackdata)
|
||||
shared.sqlSubmitQueue.put('''UPDATE sent SET lastactiontime=?, msgretrynumber=?, status=? WHERE ackdata=?''')
|
||||
shared.sqlSubmitQueue.put(t)
|
||||
shared.sqlReturnQueue.get()
|
||||
shared.workerQueue.put(('sendmessage',toaddress))
|
||||
shared.sqlSubmitQueue.put('commit')
|
||||
shared.workerQueue.put(('sendmessage',''))
|
||||
#self.emit(SIGNAL("updateStatusBar(PyQt_PyObject)"),"Doing work necessary to again attempt to deliver a message...")
|
||||
shared.UISignalQueue.put(('updateStatusBar','Doing work necessary to again attempt to deliver a message...'))
|
||||
shared.sqlSubmitQueue.put('commit')
|
||||
|
|
@ -2608,85 +2618,34 @@ class singleWorker(threading.Thread):
|
|||
threading.Thread.__init__(self)
|
||||
|
||||
def run(self):
|
||||
time.sleep(10)
|
||||
shared.sqlLock.acquire()
|
||||
shared.sqlSubmitQueue.put('''SELECT toripe FROM sent WHERE (status=? AND folder='sent')''')
|
||||
shared.sqlSubmitQueue.put(('findingpubkey',))
|
||||
shared.sqlSubmitQueue.put('''SELECT toripe FROM sent WHERE (status='awaitingpubkey' AND folder='sent')''')
|
||||
shared.sqlSubmitQueue.put('')
|
||||
queryreturn = shared.sqlReturnQueue.get()
|
||||
shared.sqlLock.release()
|
||||
for row in queryreturn:
|
||||
toripe, = row
|
||||
#It is possible for the status of a message in our sent folder (which is also our 'outbox' folder) to have a status of 'findingpubkey' even if we have the pubkey. This can
|
||||
#happen if the worker thread is working on the POW for an earlier message and does not get to the message in question before the user closes Bitmessage. In this case, the
|
||||
#status will still be 'findingpubkey' but Bitmessage will never have checked to see whether it actually already has the pubkey. We should therefore check here.
|
||||
shared.sqlLock.acquire()
|
||||
shared.sqlSubmitQueue.put('''SELECT hash FROM pubkeys WHERE hash=? ''')
|
||||
shared.sqlSubmitQueue.put((toripe,))
|
||||
queryreturn = shared.sqlReturnQueue.get()
|
||||
shared.sqlLock.release()
|
||||
if queryreturn != []: #If we have the pubkey then send the message otherwise put the hash in the neededPubkeys data structure so that we will pay attention to it if it comes over the wire.
|
||||
self.sendMsg(toripe)
|
||||
else:
|
||||
neededPubkeys[toripe] = 0
|
||||
|
||||
self.sendBroadcast() #just in case there are any proof of work tasks for Broadcasts that have yet to be sent.
|
||||
|
||||
#Now let us see if there are any proofs of work for msg messages that we have yet to complete..
|
||||
shared.sqlLock.acquire()
|
||||
t = ('doingpow',)
|
||||
shared.sqlSubmitQueue.put('''SELECT toripe FROM sent WHERE status=? and folder='sent' ''')
|
||||
shared.sqlSubmitQueue.put(t)
|
||||
shared.sqlSubmitQueue.put('''SELECT DISTINCT toaddress FROM sent WHERE (status='doingpubkeypow' AND folder='sent')''')
|
||||
shared.sqlSubmitQueue.put('')
|
||||
queryreturn = shared.sqlReturnQueue.get()
|
||||
shared.sqlLock.release()
|
||||
for row in queryreturn:
|
||||
toripe, = row
|
||||
#Evidentially there is a remote possibility that we may, for some reason, no longer have the recipient's pubkey. Let us make sure we still have it or else the sendMsg function will appear to freeze.
|
||||
shared.sqlLock.acquire()
|
||||
shared.sqlSubmitQueue.put('''SELECT hash FROM pubkeys WHERE hash=? ''')
|
||||
shared.sqlSubmitQueue.put((toripe,))
|
||||
queryreturn = shared.sqlReturnQueue.get()
|
||||
shared.sqlLock.release()
|
||||
if queryreturn != []:
|
||||
#We have the needed pubkey
|
||||
self.sendMsg(toripe)
|
||||
else:
|
||||
shared.printLock.acquire()
|
||||
sys.stderr.write('For some reason, the status of a message in our outbox is \'doingpow\' even though we lack the pubkey. Here is the RIPE hash of the needed pubkey: %s\n' % toripe.encode('hex'))
|
||||
shared.printLock.release()
|
||||
toaddress, = row
|
||||
self.requestPubKey(toaddress)
|
||||
|
||||
time.sleep(10) #give some time for the GUI to start before we start on any existing POW tasks.
|
||||
|
||||
self.sendMsg() #just in case there are any pending tasks for msg messages that have yet to be sent.
|
||||
self.sendBroadcast() #just in case there are any tasks for Broadcasts that have yet to be sent.
|
||||
|
||||
while True:
|
||||
command, data = shared.workerQueue.get()
|
||||
#statusbar = 'The singleWorker thread is working on work.'
|
||||
#self.emit(SIGNAL("updateStatusBar(PyQt_PyObject)"),statusbar)
|
||||
if command == 'sendmessage':
|
||||
toAddress = data
|
||||
toStatus,toAddressVersionNumber,toStreamNumber,toRipe = decodeAddress(toAddress)
|
||||
#print 'message type', type(message)
|
||||
#print repr(message.toUtf8())
|
||||
#print str(message.toUtf8())
|
||||
shared.sqlLock.acquire()
|
||||
shared.sqlSubmitQueue.put('SELECT hash FROM pubkeys WHERE hash=?')
|
||||
shared.sqlSubmitQueue.put((toRipe,))
|
||||
queryreturn = shared.sqlReturnQueue.get()
|
||||
shared.sqlLock.release()
|
||||
#print 'queryreturn', queryreturn
|
||||
if queryreturn == []:
|
||||
#We'll need to request the pub key because we don't have it.
|
||||
if not toRipe in neededPubkeys:
|
||||
neededPubkeys[toRipe] = 0
|
||||
print 'requesting pubkey:', toRipe.encode('hex')
|
||||
self.requestPubKey(toAddressVersionNumber,toStreamNumber,toRipe)
|
||||
else:
|
||||
print 'We have already requested this pubkey (the ripe hash is in neededPubkeys). We will re-request again soon.'
|
||||
#self.emit(SIGNAL("updateSentItemStatusByHash(PyQt_PyObject,PyQt_PyObject)"),toRipe,'Public key was requested earlier. Receiver must be offline. Will retry.')
|
||||
shared.UISignalQueue.put(('updateSentItemStatusByHash',(toRipe,'Public key was requested earlier. Receiver must be offline. Will retry.')))
|
||||
|
||||
else:
|
||||
print 'We already have the necessary public key.'
|
||||
self.sendMsg(toRipe) #by calling this function, we are asserting that we already have the pubkey for toRipe
|
||||
self.sendMsg()
|
||||
elif command == 'sendbroadcast':
|
||||
print 'Within WorkerThread, processing sendbroadcast command.'
|
||||
fromAddress,subject,message = data
|
||||
self.sendBroadcast()
|
||||
elif command == 'doPOWForMyV2Pubkey':
|
||||
self.doPOWForMyV2Pubkey(data)
|
||||
|
|
@ -2697,7 +2656,14 @@ class singleWorker(threading.Thread):
|
|||
if toRipe in neededPubkeys:
|
||||
print 'We have been awaiting the arrival of this pubkey.'
|
||||
del neededPubkeys[toRipe]
|
||||
self.sendMsg(toRipe)
|
||||
t = (toRipe,)
|
||||
shared.sqlLock.acquire()
|
||||
shared.sqlSubmitQueue.put('''UPDATE sent SET status='doingmsgpow' WHERE toripe=? AND status='awaitingpubkey' and folder='sent' ''')
|
||||
shared.sqlSubmitQueue.put(t)
|
||||
shared.sqlReturnQueue.get()
|
||||
shared.sqlSubmitQueue.put('commit')
|
||||
shared.sqlLock.release()
|
||||
self.sendMsg()
|
||||
else:
|
||||
shared.printLock.acquire()
|
||||
print 'We don\'t need this pub key. We didn\'t ask for it. Pubkey hash:', toRipe.encode('hex')
|
||||
|
|
@ -2706,7 +2672,6 @@ class singleWorker(threading.Thread):
|
|||
shared.printLock.acquire()
|
||||
sys.stderr.write('Probable programming error: The command sent to the workerThread is weird. It is: %s\n' % command)
|
||||
shared.printLock.release()
|
||||
|
||||
shared.workerQueue.task_done()
|
||||
|
||||
def doPOWForMyV2Pubkey(self,hash): #This function also broadcasts out the pubkey message once it is done with the POW
|
||||
|
|
@ -2902,7 +2867,7 @@ class singleWorker(threading.Thread):
|
|||
inventoryHash = calculateInventoryHash(payload)
|
||||
objectType = 'broadcast'
|
||||
shared.inventory[inventoryHash] = (objectType, streamNumber, payload, int(time.time()))
|
||||
print 'sending inv (within sendBroadcast function)'
|
||||
print 'Broadcasting inv for my broadcast (within sendBroadcast function):', inventoryHash.encode('hex')
|
||||
shared.broadcastToSendDataQueues((streamNumber, 'sendinv', inventoryHash))
|
||||
|
||||
#self.emit(SIGNAL("updateSentItemStatusByAckdata(PyQt_PyObject,PyQt_PyObject)"),ackdata,'Broadcast sent on '+unicode(strftime(shared.config.get('bitmessagesettings', 'timeformat'),localtime(int(time.time()))),'utf-8'))
|
||||
|
|
@ -2992,25 +2957,83 @@ class singleWorker(threading.Thread):
|
|||
sys.stderr.write('Error: In the singleWorker thread, the sendBroadcast function doesn\'t understand the address version.\n')
|
||||
shared.printLock.release()
|
||||
|
||||
def sendMsg(self,toRipe):
|
||||
def sendMsg(self):
|
||||
#Check to see if there are any messages queued to be sent
|
||||
shared.sqlLock.acquire()
|
||||
t = ('doingpow','findingpubkey',toRipe)
|
||||
shared.sqlSubmitQueue.put('''UPDATE sent SET status=? WHERE status=? AND toripe=? and folder='sent' ''')
|
||||
shared.sqlSubmitQueue.put(t)
|
||||
shared.sqlSubmitQueue.put('''SELECT toaddress FROM sent WHERE (status='msgqueued' AND folder='sent')''')
|
||||
shared.sqlSubmitQueue.put('')
|
||||
queryreturn = shared.sqlReturnQueue.get()
|
||||
shared.sqlSubmitQueue.put('commit')
|
||||
|
||||
t = ('doingpow',toRipe)
|
||||
shared.sqlSubmitQueue.put('''SELECT toaddress, fromaddress, subject, message, ackdata FROM sent WHERE status=? AND toripe=? and folder='sent' ''')
|
||||
shared.sqlLock.release()
|
||||
for row in queryreturn: #For each address to which we need to send a message, check to see if we have its pubkey already.
|
||||
toaddress, = row
|
||||
toripe = decodeAddress(toaddress)[3]
|
||||
shared.sqlLock.acquire()
|
||||
shared.sqlSubmitQueue.put('''SELECT hash FROM pubkeys WHERE hash=? ''')
|
||||
shared.sqlSubmitQueue.put((toripe,))
|
||||
queryreturn = shared.sqlReturnQueue.get()
|
||||
shared.sqlLock.release()
|
||||
if queryreturn != []: #If we have the needed pubkey, set the status to doingmsgpow (we'll do it further down)
|
||||
t = (toaddress,)
|
||||
shared.sqlLock.acquire()
|
||||
shared.sqlSubmitQueue.put('''UPDATE sent SET status='doingmsgpow' WHERE toaddress=? AND status='msgqueued' ''')
|
||||
shared.sqlSubmitQueue.put(t)
|
||||
shared.sqlReturnQueue.get()
|
||||
shared.sqlSubmitQueue.put('commit')
|
||||
shared.sqlLock.release()
|
||||
else: #We don't have the needed pubkey. Set the status to 'awaitingpubkey' and request it if we haven't already
|
||||
if toripe in neededPubkeys:
|
||||
#We already sent a request for the pubkey
|
||||
t = (toaddress,)
|
||||
shared.sqlLock.acquire()
|
||||
shared.sqlSubmitQueue.put('''UPDATE sent SET status='awaitingpubkey' WHERE toaddress=? AND status='msgqueued' ''')
|
||||
shared.sqlSubmitQueue.put(t)
|
||||
shared.sqlReturnQueue.get()
|
||||
shared.sqlSubmitQueue.put('commit')
|
||||
shared.sqlLock.release()
|
||||
shared.UISignalQueue.put(('updateSentItemStatusByHash',(toripe,'Encryption key was requested earlier.')))
|
||||
else:
|
||||
#We have not yet sent a request for the pubkey
|
||||
t = (toaddress,)
|
||||
shared.sqlLock.acquire()
|
||||
shared.sqlSubmitQueue.put('''UPDATE sent SET status='doingpubkeypow' WHERE toaddress=? AND status='msgqueued' ''')
|
||||
shared.sqlSubmitQueue.put(t)
|
||||
shared.sqlReturnQueue.get()
|
||||
shared.sqlSubmitQueue.put('commit')
|
||||
shared.sqlLock.release()
|
||||
shared.UISignalQueue.put(('updateSentItemStatusByHash',(toripe,'Sending a request for the recipient\'s encryption key.')))
|
||||
self.requestPubKey(toaddress)
|
||||
shared.sqlLock.acquire()
|
||||
shared.sqlSubmitQueue.put('''SELECT toaddress, toripe, fromaddress, subject, message, ackdata FROM sent WHERE status='doingmsgpow' and folder='sent' ''')
|
||||
shared.sqlSubmitQueue.put('')
|
||||
queryreturn = shared.sqlReturnQueue.get()
|
||||
shared.sqlLock.release()
|
||||
for row in queryreturn:
|
||||
toaddress, fromaddress, subject, message, ackdata = row
|
||||
toaddress, toripe, fromaddress, subject, message, ackdata = row
|
||||
|
||||
#Evidently there is a remote possibility that we may no longer have the recipient's pubkey. Let us make sure we still have it or else the sendMsg function will appear to freeze.
|
||||
shared.sqlLock.acquire()
|
||||
shared.sqlSubmitQueue.put('''SELECT hash FROM pubkeys WHERE hash=? ''')
|
||||
shared.sqlSubmitQueue.put((toripe,))
|
||||
queryreturn = shared.sqlReturnQueue.get()
|
||||
shared.sqlLock.release()
|
||||
if queryreturn == []:
|
||||
#We no longer have the needed pubkey
|
||||
shared.printLock.acquire()
|
||||
sys.stderr.write('For some reason, the status of a message in our outbox is \'doingmsgpow\' even though we lack the pubkey. Here is the RIPE hash of the needed pubkey: %s\n' % toripe.encode('hex'))
|
||||
shared.printLock.release()
|
||||
t = (toaddress,)
|
||||
shared.sqlLock.acquire()
|
||||
shared.sqlSubmitQueue.put('''UPDATE sent SET status='msgqueued' WHERE toaddress=? AND status='doingmsgpow' ''')
|
||||
shared.sqlSubmitQueue.put(t)
|
||||
shared.sqlReturnQueue.get()
|
||||
shared.sqlSubmitQueue.put('commit')
|
||||
shared.sqlLock.release()
|
||||
shared.UISignalQueue.put(('updateSentItemStatusByHash',(toripe,'Sending a request for the recipient\'s encryption key.')))
|
||||
self.requestPubKey(toaddress)
|
||||
return
|
||||
ackdataForWhichImWatching[ackdata] = 0
|
||||
toStatus,toAddressVersionNumber,toStreamNumber,toHash = decodeAddress(toaddress)
|
||||
fromStatus,fromAddressVersionNumber,fromStreamNumber,fromHash = decodeAddress(fromaddress)
|
||||
#self.emit(SIGNAL("updateSentItemStatusByAckdata(PyQt_PyObject,PyQt_PyObject)"),ackdata,'Doing work necessary to send the message.')
|
||||
shared.UISignalQueue.put(('updateSentItemStatusByAckdata',(ackdata,'Doing work necessary to send the message.')))
|
||||
shared.printLock.acquire()
|
||||
print 'Found a message in our database that needs to be sent with this pubkey.'
|
||||
|
|
@ -3102,7 +3125,7 @@ class singleWorker(threading.Thread):
|
|||
if toAddressVersionNumber == 2 or toAddressVersionNumber == 3:
|
||||
shared.sqlLock.acquire()
|
||||
shared.sqlSubmitQueue.put('SELECT transmitdata FROM pubkeys WHERE hash=?')
|
||||
shared.sqlSubmitQueue.put((toRipe,))
|
||||
shared.sqlSubmitQueue.put((toripe,))
|
||||
queryreturn = shared.sqlReturnQueue.get()
|
||||
shared.sqlLock.release()
|
||||
if queryreturn == []:
|
||||
|
|
@ -3164,17 +3187,17 @@ class singleWorker(threading.Thread):
|
|||
shared.inventory[inventoryHash] = (objectType, toStreamNumber, payload, int(time.time()))
|
||||
#self.emit(SIGNAL("updateSentItemStatusByAckdata(PyQt_PyObject,PyQt_PyObject)"),ackdata,'Message sent. Waiting on acknowledgement. Sent on ' + unicode(strftime(shared.config.get('bitmessagesettings', 'timeformat'),localtime(int(time.time()))),'utf-8'))
|
||||
shared.UISignalQueue.put(('updateSentItemStatusByAckdata',(ackdata,'Message sent. Waiting on acknowledgement. Sent on ' + unicode(strftime(shared.config.get('bitmessagesettings', 'timeformat'),localtime(int(time.time()))),'utf-8'))))
|
||||
print 'sending inv (within sendmsg function)'
|
||||
print 'Broadcasting inv for my msg(within sendmsg function):', inventoryHash.encode('hex')
|
||||
shared.broadcastToSendDataQueues((streamNumber, 'sendinv', inventoryHash))
|
||||
|
||||
#Update the status of the message in the 'sent' table to have a 'sent' status
|
||||
shared.sqlLock.acquire()
|
||||
t = ('sentmessage',toaddress, fromaddress, subject, message,'doingpow')
|
||||
t = ('msgsent',toaddress, fromaddress, subject, message,'doingmsgpow')
|
||||
shared.sqlSubmitQueue.put('UPDATE sent SET status=? WHERE toaddress=? AND fromaddress=? AND subject=? AND message=? AND status=?')
|
||||
shared.sqlSubmitQueue.put(t)
|
||||
queryreturn = shared.sqlReturnQueue.get()
|
||||
|
||||
t = (toRipe,)
|
||||
t = (toripe,)
|
||||
shared.sqlSubmitQueue.put('''UPDATE pubkeys SET usedpersonally='yes' WHERE hash=?''')
|
||||
shared.sqlSubmitQueue.put(t)
|
||||
queryreturn = shared.sqlReturnQueue.get()
|
||||
|
|
@ -3182,7 +3205,9 @@ class singleWorker(threading.Thread):
|
|||
shared.sqlLock.release()
|
||||
|
||||
|
||||
def requestPubKey(self,addressVersionNumber,streamNumber,ripe):
|
||||
def requestPubKey(self,toAddress):
|
||||
toStatus,addressVersionNumber,streamNumber,ripe = decodeAddress(toAddress)
|
||||
neededPubkeys[ripe] = 0
|
||||
payload = pack('>I',(int(time.time())+random.randrange(-300, 300)))#the current time plus or minus five minutes.
|
||||
payload += encodeVarint(addressVersionNumber)
|
||||
payload += encodeVarint(streamNumber)
|
||||
|
|
@ -3194,10 +3219,8 @@ class singleWorker(threading.Thread):
|
|||
trialValue = 99999999999999999999
|
||||
#print 'trial value', trialValue
|
||||
statusbar = 'Doing the computations necessary to request the recipient\'s public key.'
|
||||
#self.emit(SIGNAL("updateStatusBar(PyQt_PyObject)"),statusbar)
|
||||
shared.UISignalQueue.put(('updateStatusBar',statusbar))
|
||||
#self.emit(SIGNAL("updateSentItemStatusByHash(PyQt_PyObject,PyQt_PyObject)"),ripe,'Doing work necessary to request public key.')
|
||||
shared.UISignalQueue.put(('updateSentItemStatusByHash',(ripe,'Doing work necessary to request public key.')))
|
||||
shared.UISignalQueue.put(('updateSentItemStatusByHash',(ripe,'Doing work necessary to request encryption key.')))
|
||||
print 'Doing proof-of-work necessary to send getpubkey message.'
|
||||
target = 2**64 / ((len(payload)+shared.networkDefaultPayloadLengthExtraBytes+8) * shared.networkDefaultProofOfWorkNonceTrialsPerByte)
|
||||
initialHash = hashlib.sha512(payload).digest()
|
||||
|
|
@ -3215,9 +3238,15 @@ class singleWorker(threading.Thread):
|
|||
print 'sending inv (for the getpubkey message)'
|
||||
shared.broadcastToSendDataQueues((streamNumber, 'sendinv', inventoryHash))
|
||||
|
||||
#self.emit(SIGNAL("updateStatusBar(PyQt_PyObject)"),'Broacasting the public key request. This program will auto-retry if they are offline.')
|
||||
t = (toAddress,)
|
||||
shared.sqlLock.acquire()
|
||||
shared.sqlSubmitQueue.put('''UPDATE sent SET status='awaitingpubkey' WHERE toaddress=? AND status='doingpubkeypow' ''')
|
||||
shared.sqlSubmitQueue.put(t)
|
||||
shared.sqlReturnQueue.get()
|
||||
shared.sqlSubmitQueue.put('commit')
|
||||
shared.sqlLock.release()
|
||||
|
||||
shared.UISignalQueue.put(('updateStatusBar','Broacasting the public key request. This program will auto-retry if they are offline.'))
|
||||
#self.emit(SIGNAL("updateSentItemStatusByHash(PyQt_PyObject,PyQt_PyObject)"),ripe,'Sending public key request. Waiting for reply. Requested at ' + unicode(strftime(shared.config.get('bitmessagesettings', 'timeformat'),localtime(int(time.time()))),'utf-8'))
|
||||
shared.UISignalQueue.put(('updateSentItemStatusByHash',(ripe,'Sending public key request. Waiting for reply. Requested at ' + unicode(strftime(shared.config.get('bitmessagesettings', 'timeformat'),localtime(int(time.time()))),'utf-8'))))
|
||||
|
||||
def generateFullAckMessage(self,ackdata,toStreamNumber,embeddedTime):
|
||||
|
|
@ -3725,7 +3754,7 @@ class MySimpleXMLRPCRequestHandler(SimpleXMLRPCRequestHandler):
|
|||
|
||||
ackdata = OpenSSL.rand(32)
|
||||
shared.sqlLock.acquire()
|
||||
t = ('',toAddress,toRipe,fromAddress,subject,message,ackdata,int(time.time()),'findingpubkey',1,1,'sent',2)
|
||||
t = ('',toAddress,toRipe,fromAddress,subject,message,ackdata,int(time.time()),'msgqueued',1,1,'sent',2)
|
||||
shared.sqlSubmitQueue.put('''INSERT INTO sent VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)''')
|
||||
shared.sqlSubmitQueue.put(t)
|
||||
shared.sqlReturnQueue.get()
|
||||
|
|
@ -3799,7 +3828,7 @@ class MySimpleXMLRPCRequestHandler(SimpleXMLRPCRequestHandler):
|
|||
#apiSignalQueue.put(('displayNewSentMessage',(toAddress,toLabel,fromAddress,subject,message,ackdata)))
|
||||
#self.emit(SIGNAL("displayNewSentMessage(PyQt_PyObject,PyQt_PyObject,PyQt_PyObject,PyQt_PyObject,PyQt_PyObject,PyQt_PyObject)"),toAddress,toLabel,fromAddress,subject,message,ackdata)
|
||||
shared.UISignalQueue.put(('displayNewSentMessage',(toAddress,toLabel,fromAddress,subject,message,ackdata)))
|
||||
shared.workerQueue.put(('sendbroadcast',(fromAddress,subject,message)))
|
||||
shared.workerQueue.put(('sendbroadcast',''))
|
||||
|
||||
return ackdata.encode('hex')
|
||||
elif method == 'getStatus':
|
||||
|
|
@ -3817,16 +3846,7 @@ class MySimpleXMLRPCRequestHandler(SimpleXMLRPCRequestHandler):
|
|||
return 'notFound'
|
||||
for row in queryreturn:
|
||||
status, = row
|
||||
if status == 'findingpubkey':
|
||||
return 'findingPubkey'
|
||||
if status == 'doingpow':
|
||||
return 'doingPow'
|
||||
if status == 'sentmessage':
|
||||
return 'sentMessage'
|
||||
if status == 'ackreceived':
|
||||
return 'ackReceived'
|
||||
else:
|
||||
return 'otherStatus: '+status
|
||||
return status
|
||||
elif method == 'addSubscription':
|
||||
if len(params) == 0:
|
||||
return 'API Error 0000: I need parameters!'
|
||||
|
|
@ -4056,7 +4076,7 @@ if __name__ == "__main__":
|
|||
shared.reloadBroadcastSendersForWhichImWatching()
|
||||
|
||||
#Initialize the ackdataForWhichImWatching data structure using data from the sql database.
|
||||
shared.sqlSubmitQueue.put('''SELECT ackdata FROM sent where (status='sentmessage' OR status='doingpow')''')
|
||||
shared.sqlSubmitQueue.put('''SELECT ackdata FROM sent where (status='msgsent' OR status='doingmsgpow')''')
|
||||
shared.sqlSubmitQueue.put('')
|
||||
queryreturn = shared.sqlReturnQueue.get()
|
||||
for row in queryreturn:
|
||||
|
|
|
|||
|
|
@ -358,16 +358,20 @@ class MyForm(QtGui.QMainWindow):
|
|||
newItem.setData(Qt.UserRole,unicode(message,'utf-8)'))
|
||||
newItem.setFlags( QtCore.Qt.ItemIsSelectable | QtCore.Qt.ItemIsEnabled )
|
||||
self.ui.tableWidgetSent.setItem(0,2,newItem)
|
||||
if status == 'findingpubkey':
|
||||
newItem = myTableWidgetItem('Waiting on their public key. Will request it again soon.')
|
||||
elif status == 'sentmessage':
|
||||
if status == 'awaitingpubkey':
|
||||
newItem = myTableWidgetItem('Waiting on their encryption key. Will request it again soon.')
|
||||
elif status == 'doingpowforpubkey':
|
||||
newItem = myTableWidgetItem('Encryption key request queued.')
|
||||
elif status == 'msgqueued':
|
||||
newItem = myTableWidgetItem('Queued.')
|
||||
elif status == 'msgsent':
|
||||
newItem = myTableWidgetItem('Message sent. Waiting on acknowledgement. Sent at ' + unicode(strftime(shared.config.get('bitmessagesettings', 'timeformat'),localtime(lastactiontime)),'utf-8'))
|
||||
elif status == 'doingpow':
|
||||
elif status == 'doingmsgpow':
|
||||
newItem = myTableWidgetItem('Need to do work to send message. Work is queued.')
|
||||
elif status == 'ackreceived':
|
||||
newItem = myTableWidgetItem('Acknowledgement of the message received ' + unicode(strftime(shared.config.get('bitmessagesettings', 'timeformat'),localtime(int(lastactiontime))),'utf-8'))
|
||||
elif status == 'broadcastpending':
|
||||
newItem = myTableWidgetItem('Doing the work necessary to send broadcast...')
|
||||
elif status == 'broadcastqueued':
|
||||
newItem = myTableWidgetItem('Broadcast queued.')
|
||||
elif status == 'broadcastsent':
|
||||
newItem = myTableWidgetItem('Broadcast on ' + unicode(strftime(shared.config.get('bitmessagesettings', 'timeformat'),localtime(int(lastactiontime))),'utf-8'))
|
||||
else:
|
||||
|
|
@ -770,6 +774,7 @@ class MyForm(QtGui.QMainWindow):
|
|||
def click_actionDeleteAllTrashedMessages(self):
|
||||
if QtGui.QMessageBox.question(self, 'Delete trash?',"Are you sure you want to delete all trashed messages?", QtGui.QMessageBox.Yes, QtGui.QMessageBox.No) == QtGui.QMessageBox.No:
|
||||
return
|
||||
self.statusBar().showMessage('Deleting messages and freeing empty space...')
|
||||
shared.sqlLock.acquire()
|
||||
shared.sqlSubmitQueue.put('''delete from inbox where folder='trash' ''')
|
||||
shared.sqlSubmitQueue.put('')
|
||||
|
|
@ -781,6 +786,7 @@ class MyForm(QtGui.QMainWindow):
|
|||
shared.sqlSubmitQueue.put('vacuum')
|
||||
shared.sqlSubmitQueue.put('')
|
||||
shared.sqlReturnQueue.get()
|
||||
self.statusBar().showMessage('')
|
||||
shared.sqlLock.release()
|
||||
|
||||
def click_actionRegenerateDeterministicAddresses(self):
|
||||
|
|
@ -1104,7 +1110,7 @@ class MyForm(QtGui.QMainWindow):
|
|||
self.statusBar().showMessage('Warning: You are currently not connected. Bitmessage will do the work necessary to send the message but it won\'t send until you connect.')
|
||||
ackdata = OpenSSL.rand(32)
|
||||
shared.sqlLock.acquire()
|
||||
t = ('',toAddress,ripe,fromAddress,subject,message,ackdata,int(time.time()),'findingpubkey',1,1,'sent',2)
|
||||
t = ('',toAddress,ripe,fromAddress,subject,message,ackdata,int(time.time()),'msgqueued',1,1,'sent',2)
|
||||
shared.sqlSubmitQueue.put('''INSERT INTO sent VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)''')
|
||||
shared.sqlSubmitQueue.put(t)
|
||||
shared.sqlReturnQueue.get()
|
||||
|
|
@ -1151,7 +1157,7 @@ class MyForm(QtGui.QMainWindow):
|
|||
shared.sqlSubmitQueue.put('commit')
|
||||
shared.sqlLock.release()
|
||||
|
||||
shared.workerQueue.put(('sendbroadcast',(fromAddress,subject,message)))
|
||||
shared.workerQueue.put(('sendbroadcast',''))
|
||||
|
||||
try:
|
||||
fromLabel = shared.config.get(fromAddress, 'label')
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user