<commit_before>from twisted.protocols import stateful
from twisted.internet import reactor
from twisted.internet.protocol import Factory, Protocol
from twisted.internet.endpoints import TCP4ClientEndpoint

from datetime import datetime

import sys
import struct
import zlib
import time
import threading
import Queue
import uuid

import AmmoMessages_pb2

class AndroidProtocol(stateful.StatefulProtocol):
  '''
  This class implements the stateful Android <-> Gateway protocol.  It contains
  an 8-byte header with messageSize and a checksum, then a protobuf
  MessageWrapper object (of length messageSize).
  
  We use Twisted's StatefulProtocol to implement this protocol--  states are
  composed of a callback function and a size; Twisted calls the callback
  function when <size> data has been received.  The callback functions return
  the next state.
  '''
  _messageSize = 0
  _checksum = 0
  _onMessageAvailableCallback = None
  
  def getInitialState(self):
    return (self.receiveHeader, 8) #initial state receives the header
  
  def receiveHeader(self, data):
    (messageSize, checksum) = struct.unpack("<Ii", data)
    self._messageSize = messageSize
    self._checksum = checksum
    
    return (self.receiveData, self._messageSize)
    
  def receiveData(self, data):
    calculatedChecksum = zlib.crc32(data)
    if calculatedChecksum != self._checksum:
      print "Checksum error!"
    msg = AmmoMessages_pb2.MessageWrapper()
    msg.ParseFromString(data)
    
    if self._onMessageAvailableCallback != None:
      self._onMessageAvailableCallback(msg)
    
    return (self.receiveHeader, 8)
    
  def sendMessageWrapper(self, msg):
    serializedMsg = msg.SerializeToString()
    self.transport.write(struct.pack("<Ii", len(serializedMsg), zlib.crc32(serializedMsg))) #little-endian byte order for now
    self.transport.write(serializedMsg);
    
  def connectionMade(self):
    pass
    
  def setOnMessageAvailableCallback(self, callback):
    self._onMessageAvailableCallback = callback
    
class AuthenticationFailure(Exception):
  pass
    
    
class MessageScope:
    GLOBAL = 0
    LOCAL = 1
class AndroidConnector(threading.Thread):
  _address = ""
  _port = 0
  _deviceId = ""
  _userId = ""
  _userKey = ""
  
  _protocol = None
  
  _authenticated = False
  _authCondition = None
  
  _messageQueueEnabled = True
  _messageQueue = None
  _messageCallback = None
  
  def __init__(self, address, port, deviceId, userId, userKey):
    threading.Thread.__init__(self)
    self._address = address
    self._port = port
    self._deviceId = deviceId
    self._userId = userId
    self._userKey = userKey
    
    self._authenticated = False
    self._authCondition = threading.Condition()
    
    self._messageQueueEnabled = True
    self._messageQueue = Queue.Queue()
    self._messageCallback = None
    
  def _gotProtocol(self, p):
    self._protocol = p
    self._onConnect()
    
  def _connect(self):
    factory = Factory()
    factory.protocol = AndroidProtocol
    point = TCP4ClientEndpoint(reactor, self._address, self._port)
    d = point.connect(factory)
    d.addCallback(self._gotProtocol)
    
  def run(self):
    if reactor.running == False:
      self._connect()
      print "Running reactor"
      reactor.run(False) #Argument False tells the reactor that it's not on the
                         #main thread, so it doesn't attempt to register signal
                         #handlers (which doesn't work on other threads)
      print "Reactor stopped"
    else:
      reactor.callFromThread(self._connect)
      print "Reactor is already running...  this background thread will exit."
    
  def _onConnect(self):
    self._protocol.setOnMessageAvailableCallback(self._onMessageAvailable)
    self._sendAuthMessage()
    
  def _onMessageAvailable(self, msg):
    if self._authenticated == False:
      if msg.type == AmmoMessages_pb2.MessageWrapper.AUTHENTICATION_RESULT:
        if msg.authentication_result.result == AmmoMessages_pb2.AuthenticationResult.SUCCESS:
          print "Authentication succeeded."
          self._authCondition.acquire()
          self._authenticated = True
          self._authCondition.notifyAll()
          self._authCondition.release()
        else:
          print "Authentication failed."
          raise AuthenticationFailure("Auth failed: " + msg.authentication_result.message)
    
    time = datetime.now()
    if self._messageCallback != None:
      self._messageCallback(self, msg)
    
    if self._messageQueueEnabled:
      self._messageQueue.put((msg, time))
    
  def _sendAuthMessage(self):
    m = AmmoMessages_pb2.MessageWrapper()
    m.type = AmmoMessages_pb2.MessageWrapper.AUTHENTICATION_MESSAGE
    m.authentication_message.device_id = self._deviceId
    m.authentication_message.user_id = self._userId
    m.authentication_message.user_key = self._userKey
    print "Sending auth message"
    self._protocol.sendMessageWrapper(m)
    
  def dequeueMessage(self):
    '''
    Dequeues a message from the message queue and returns it.  Returns 'none' if
    the queue is empty; otherwise, it returns a pair (message, timeReceived).
    '''
    item = None
    try:
      item = self._messageQueue.get(False) #don't block if queue is empty; raises Empty exception instead
    except Queue.Empty:
      item = None
      pass
    
    return item
    
  def isDataAvailable(self):
    '''
    Checks to see if data is available in the message queue.  Note that, since
    the message queue is filled from a background thread (and could be emptied
    from a background thread), this method returning true/false does not
    necessarily mean that a message will or will not be present when
    dequeueMessage() is called.
    '''
    return not self._messageQueue.empty()
    
  def push(self, uri, mimeType, data):
    '''
    Sends a push message with the specified URI and MIME type to the gateway.
    '''
    m = AmmoMessages_pb2.MessageWrapper()
    m.type = AmmoMessages_pb2.MessageWrapper.DATA_MESSAGE
    m.data_message.uri = uri
    m.data_message.mime_type = mimeType
    m.data_message.data = data
    reactor.callFromThread(self._protocol.sendMessageWrapper, m)
    
  def subscribe(self, mimeType, scope = MessageScope.GLOBAL):
    '''
    Subscribes to push data with the specified MIME type.
    
    By default, data received will be placed in the message queue.  The caller
    should periodically call dequeueMessage to receive the push messages that
    it subscribed to.
    '''
    m = AmmoMessages_pb2.MessageWrapper()
    m.type = AmmoMessages_pb2.MessageWrapper.SUBSCRIBE_MESSAGE
    m.subscribe_message.mime_type = mimeType
    if scope == MessageScope.GLOBAL:
      m.subscribe_message.scope = AmmoMessages_pb2.GLOBAL
    else:
      m.subscribe_message.scope = AmmoMessages_pb2.LOCAL
    reactor.callFromThread(self._protocol.sendMessageWrapper, m)
    
  def pullRequest(self, mimeType, query, projection, maxResults, startFromCount, liveQuery):
    '''
    Sends a pull request with the specified parameters.  Note that the request<commit_msg>Add push scoping to AndroidConnector interface<commit_after>from twisted.protocols import stateful
from twisted.internet import reactor
from twisted.internet.protocol import Factory, Protocol
from twisted.internet.endpoints import TCP4ClientEndpoint

from datetime import datetime

import sys
import struct
import zlib
import time
import threading
import Queue
import uuid

import AmmoMessages_pb2

class AndroidProtocol(stateful.StatefulProtocol):
  '''
  This class implements the stateful Android <-> Gateway protocol.  It contains
  an 8-byte header with messageSize and a checksum, then a protobuf
  MessageWrapper object (of length messageSize).
  
  We use Twisted's StatefulProtocol to implement this protocol--  states are
  composed of a callback function and a size; Twisted calls the callback
  function when <size> data has been received.  The callback functions return
  the next state.
  '''
  _messageSize = 0
  _checksum = 0
  _onMessageAvailableCallback = None
  
  def getInitialState(self):
    return (self.receiveHeader, 8) #initial state receives the header
  
  def receiveHeader(self, data):
    (messageSize, checksum) = struct.unpack("<Ii", data)
    self._messageSize = messageSize
    self._checksum = checksum
    
    return (self.receiveData, self._messageSize)
    
  def receiveData(self, data):
    calculatedChecksum = zlib.crc32(data)
    if calculatedChecksum != self._checksum:
      print "Checksum error!"
    msg = AmmoMessages_pb2.MessageWrapper()
    msg.ParseFromString(data)
    
    if self._onMessageAvailableCallback != None:
      self._onMessageAvailableCallback(msg)
    
    return (self.receiveHeader, 8)
    
  def sendMessageWrapper(self, msg):
    serializedMsg = msg.SerializeToString()
    self.transport.write(struct.pack("<Ii", len(serializedMsg), zlib.crc32(serializedMsg))) #little-endian byte order for now
    self.transport.write(serializedMsg);
    
  def connectionMade(self):
    pass
    
  def setOnMessageAvailableCallback(self, callback):
    self._onMessageAvailableCallback = callback
    
class AuthenticationFailure(Exception):
  pass
    
    
class MessageScope:
    GLOBAL = 0
    LOCAL = 1
class AndroidConnector(threading.Thread):
  _address = ""
  _port = 0
  _deviceId = ""
  _userId = ""
  _userKey = ""
  
  _protocol = None
  
  _authenticated = False
  _authCondition = None
  
  _messageQueueEnabled = True
  _messageQueue = None
  _messageCallback = None
  
  def __init__(self, address, port, deviceId, userId, userKey):
    threading.Thread.__init__(self)
    self._address = address
    self._port = port
    self._deviceId = deviceId
    self._userId = userId
    self._userKey = userKey
    
    self._authenticated = False
    self._authCondition = threading.Condition()
    
    self._messageQueueEnabled = True
    self._messageQueue = Queue.Queue()
    self._messageCallback = None
    
  def _gotProtocol(self, p):
    self._protocol = p
    self._onConnect()
    
  def _connect(self):
    factory = Factory()
    factory.protocol = AndroidProtocol
    point = TCP4ClientEndpoint(reactor, self._address, self._port)
    d = point.connect(factory)
    d.addCallback(self._gotProtocol)
    
  def run(self):
    if reactor.running == False:
      self._connect()
      print "Running reactor"
      reactor.run(False) #Argument False tells the reactor that it's not on the
                         #main thread, so it doesn't attempt to register signal
                         #handlers (which doesn't work on other threads)
      print "Reactor stopped"
    else:
      reactor.callFromThread(self._connect)
      print "Reactor is already running...  this background thread will exit."
    
  def _onConnect(self):
    self._protocol.setOnMessageAvailableCallback(self._onMessageAvailable)
    self._sendAuthMessage()
    
  def _onMessageAvailable(self, msg):
    if self._authenticated == False:
      if msg.type == AmmoMessages_pb2.MessageWrapper.AUTHENTICATION_RESULT:
        if msg.authentication_result.result == AmmoMessages_pb2.AuthenticationResult.SUCCESS:
          print "Authentication succeeded."
          self._authCondition.acquire()
          self._authenticated = True
          self._authCondition.notifyAll()
          self._authCondition.release()
        else:
          print "Authentication failed."
          raise AuthenticationFailure("Auth failed: " + msg.authentication_result.message)
    
    time = datetime.now()
    if self._messageCallback != None:
      self._messageCallback(self, msg)
    
    if self._messageQueueEnabled:
      self._messageQueue.put((msg, time))
    
  def _sendAuthMessage(self):
    m = AmmoMessages_pb2.MessageWrapper()
    m.type = AmmoMessages_pb2.MessageWrapper.AUTHENTICATION_MESSAGE
    m.authentication_message.device_id = self._deviceId
    m.authentication_message.user_id = self._userId
    m.authentication_message.user_key = self._userKey
    print "Sending auth message"
    self._protocol.sendMessageWrapper(m)
    
  def dequeueMessage(self):
    '''
    Dequeues a message from the message queue and returns it.  Returns 'none' if
    the queue is empty; otherwise, it returns a pair (message, timeReceived).
    '''
    item = None
    try:
      item = self._messageQueue.get(False) #don't block if queue is empty; raises Empty exception instead
    except Queue.Empty:
      item = None
      pass
    
    return item
    
  def isDataAvailable(self):
    '''
    Checks to see if data is available in the message queue.  Note that, since
    the message queue is filled from a background thread (and could be emptied
    from a background thread), this method returning true/false does not
    necessarily mean that a message will or will not be present when
    dequeueMessage() is called.
    '''
    return not self._messageQueue.empty()
    
  def push(self, uri, mimeType, data, scope = MessageScope.GLOBAL):
    '''
    Sends a push message with the specified URI and MIME type to the gateway.
    '''
    m = AmmoMessages_pb2.MessageWrapper()
    m.type = AmmoMessages_pb2.MessageWrapper.DATA_MESSAGE
    m.data_message.uri = uri
    m.data_message.mime_type = mimeType
    m.data_message.data = data
    if scope == MessageScope.GLOBAL:
      m.data_message.scope = AmmoMessages_pb2.GLOBAL
    else:
      m.data_message.scope = AmmoMessages_pb2.LOCAL
    reactor.callFromThread(self._protocol.sendMessageWrapper, m)
    
  def subscribe(self, mimeType, scope = MessageScope.GLOBAL):
    '''
    Subscribes to push data with the specified MIME type.
    
    By default, data received will be placed in the message queue.  The caller
    should periodically call dequeueMessage to receive the push messages that
    it subscribed to.
    '''
    m = AmmoMessages_pb2.MessageWrapper()
    m.type = AmmoMessages_pb2.MessageWrapper.SUBSCRIBE_MESSAGE
    m.subscribe_message.mime_type = mimeType
    if scope == MessageScope.GLOBAL:
      m.subscribe_message.scope = AmmoMessages_pb2.GLOBAL
    else:
      m.subscribe_message.scope = AmmoMessages_pb2.LOCAL
    reactor.callFromThread(self._protocol.sendMessageWrapper, m)
    
  def pullRequest(self, mimeType, query, projection, maxResults, startFromCount, liveQuery):
    '''
    Sends a pull request with the specified parameters.  Note that the request
{'fraction_non_alphanumeric': 0.07650577270830436, 'fraction_numerical': 0.0037557379329531226, 'mean_word_length': 3.271836007130125, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2645955', 'n_tokens_mistral': 4091, 'n_tokens_neox': 4074, 'n_words': 1160}