import ConfigParser
from json import dumps, loads
import urllib
-
+import threading
config = ConfigParser.ConfigParser()
# set some defaults, which will be overwritten by the config file
password = config.get('server','password')
-
stopping = False
-block_number = -1
sessions = {}
-m_sessions = [{}] # served by http
-
-peer_list = {}
-
-from Queue import Queue
-input_queue = Queue()
-output_queue = Queue()
-
-
def random_string(N):
import random, string
return ''.join(random.choice(string.ascii_uppercase + string.digits) for x in range(N))
-
-
-def cmd_stop(_,__,pw):
- global stopping
- if password == pw:
- stopping = True
- return 'ok'
- else:
- return 'wrong password'
-
-def cmd_load(_,__,pw):
- if password == pw:
- return repr( len(sessions) )
- else:
- return 'wrong password'
-
-
-
def modified_addresses(a_session):
sessions[session_id]['last_time'] = time.time()
ret, addresses = modified_addresses(session)
if ret: sessions[session_id]['addresses'] = addresses
- return repr( (block_number,ret))
-
-
-def poll_session_json(session_id, message_id):
- session = m_sessions[0].get(session_id)
- if session is None:
- raise BaseException("session not found %s"%session_id)
- else:
- m_sessions[0][session_id]['last_time'] = time.time()
- out = []
- ret, addresses = modified_addresses(session)
- if ret:
- m_sessions[0][session_id]['addresses'] = addresses
- for addr in ret:
- msg_id, status = addresses[addr]
- out.append( { 'id':msg_id, 'result':status } )
-
- msg_id, last_nb = session.get('numblocks')
- if last_nb:
- if last_nb != block_number:
- m_sessions[0][session_id]['numblocks'] = msg_id, block_number
- out.append( {'id':msg_id, 'result':block_number} )
-
- return out
-
-
-
+ return repr( (store.block_number,ret))
-def address_get_history_json(_,message_id,address):
- return store.get_history(address)
-
-def subscribe_to_numblocks_json(session_id, message_id):
- global m_sessions
- m_sessions[0][session_id]['numblocks'] = message_id,block_number
- return block_number
-
-def add_address_to_session_json(session_id, message_id, address):
- global m_sessions
- sessions = m_sessions[0]
- status = store.get_status(address)
- sessions[session_id]['addresses'][address] = (message_id, status)
- sessions[session_id]['last_time'] = time.time()
- m_sessions[0] = sessions
- return status
def add_address_to_session(session_id, address):
status = store.get_status(address)
sessions[session_id]['last_time'] = time.time()
return status
+
def new_session(version, addresses):
session_id = random_string(10)
sessions[session_id] = { 'addresses':{}, 'version':version }
return out
-def client_version_json(session_id, _, version):
- global m_sessions
- sessions = m_sessions[0]
- sessions[session_id]['version'] = version
- m_sessions[0] = sessions
-
-def create_session_json(_, __):
- sessions = m_sessions[0]
- session_id = random_string(10)
- print "creating session", session_id
- sessions[session_id] = { 'addresses':{}, 'numblocks':('','') }
- sessions[session_id]['last_time'] = time.time()
- m_sessions[0] = sessions
- return session_id
-
-
-
-def get_banner(_,__):
- return config.get('server','banner').replace('\\n','\n')
-
def update_session(session_id,addresses):
"""deprecated in 0.42, wad replaced by add_address_to_session"""
sessions[session_id]['addresses'] = {}
sessions[session_id]['last_time'] = time.time()
return 'ok'
+
def native_server_thread():
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
out = poll_session(data)
elif cmd == 'h':
- # history
address = data
out = repr( store.get_history( address ) )
- elif cmd == 'load':
- out = cmd_load(None,None,data)
-
elif cmd =='tx':
out = store.send_tx(data)
print timestr(), "sent tx:", ipaddr, out
- elif cmd == 'stop':
- out = cmd_stop(data)
-
elif cmd == 'peers':
- out = repr(peer_list.values())
+ out = repr(irc.get_peers())
else:
out = None
####################################################################
-import stratum
+from processor import Shared, Processor, Dispatcher
+from stratum_http import HttpServer
+from stratum import TcpServer
-class AbeProcessor(stratum.Processor):
- def process(self,session,request):
+class AbeProcessor(Processor):
+ def process(self,request):
message_id = request['id']
method = request['method']
params = request.get('params',[])
result = ''
if method == 'numblocks.subscribe':
- session.subscribe_to_numblocks(message_id)
- result = block_number
+ result = store.block_number
elif method == 'address.subscribe':
address = params[0]
+ store.watch_address(address)
status = store.get_status(address)
- session.subscribe_to_address(address,message_id,status)
result = status
elif method == 'client.version':
- session.version = params[0]
+ #session.version = params[0]
+ pass
elif method == 'server.banner':
result = config.get('server','banner').replace('\\n','\n')
elif method == 'server.peers':
- result = peer_list.values()
+ result = irc.get_peers()
elif method == 'address.get_history':
address = params[0]
result = store.get_history( address )
print "unknown method", request
if result!='':
- response = { 'id':message_id, 'result':result }
- self.push_response(session,response)
+ response = { 'id':message_id, 'method':method, 'params':params, 'result':result }
+ self.push_response(response)
def get_status(self,addr):
return store.get_status(addr)
+class Irc(threading.Thread):
+
+ def __init__(self, processor):
+ self.processor = processor
+ threading.Thread.__init__(self)
+ self.daemon = True
+ self.peers = {}
+
+ def get_peers(self):
+ return self.peers.values()
+
+ def run(self):
+ NICK = 'E_'+random_string(10)
+ while not self.processor.shared.stopped():
+ try:
+ s = socket.socket()
+ s.connect(('irc.freenode.net', 6667))
+ s.send('USER electrum 0 * :'+config.get('server','host')+' '+config.get('server','ircname')+'\n')
+ s.send('NICK '+NICK+'\n')
+ s.send('JOIN #electrum\n')
+ sf = s.makefile('r', 0)
+ t = 0
+ while not self.processor.shared.stopped():
+ line = sf.readline()
+ line = line.rstrip('\r\n')
+ line = line.split()
+ if line[0]=='PING':
+ s.send('PONG '+line[1]+'\n')
+ elif '353' in line: # answer to /names
+ k = line.index('353')
+ for item in line[k+1:]:
+ if item[0:2] == 'E_':
+ s.send('WHO %s\n'%item)
+ elif '352' in line: # answer to /who
+ # warning: this is a horrible hack which apparently works
+ k = line.index('352')
+ ip = line[k+4]
+ ip = socket.gethostbyname(ip)
+ name = line[k+6]
+ host = line[k+9]
+ self.peers[name] = (ip,host)
+ if time.time() - t > 5*60:
+ self.processor.push_response({'method':'server.peers', 'result':[self.get_peers()]})
+ s.send('NAMES #electrum\n')
+ t = time.time()
+ self.peers = {}
+ except:
+ traceback.print_exc(file=sys.stdout)
+ finally:
+ sf.close()
+ s.close()
-def irc_thread():
- global peer_list
- NICK = 'E_'+random_string(10)
- while not stopping:
- try:
- s = socket.socket()
- s.connect(('irc.freenode.net', 6667))
- s.send('USER electrum 0 * :'+config.get('server','host')+' '+config.get('server','ircname')+'\n')
- s.send('NICK '+NICK+'\n')
- s.send('JOIN #electrum\n')
- sf = s.makefile('r', 0)
- t = 0
- while not stopping:
- line = sf.readline()
- line = line.rstrip('\r\n')
- line = line.split()
- if line[0]=='PING':
- s.send('PONG '+line[1]+'\n')
- elif '353' in line: # answer to /names
- k = line.index('353')
- for item in line[k+1:]:
- if item[0:2] == 'E_':
- s.send('WHO %s\n'%item)
- elif '352' in line: # answer to /who
- # warning: this is a horrible hack which apparently works
- k = line.index('352')
- ip = line[k+4]
- ip = socket.gethostbyname(ip)
- name = line[k+6]
- host = line[k+9]
- peer_list[name] = (ip,host)
- if time.time() - t > 5*60:
- s.send('NAMES #electrum\n')
- t = time.time()
- peer_list = {}
- except:
- traceback.print_exc(file=sys.stdout)
- finally:
- sf.close()
- s.close()
-
-
-def get_peers_json(_,__):
- return peer_list.values()
-
-def http_server_thread():
- # see http://code.google.com/p/jsonrpclib/
- from SocketServer import ThreadingMixIn
- from StratumJSONRPCServer import StratumJSONRPCServer
- class StratumThreadedJSONRPCServer(ThreadingMixIn, StratumJSONRPCServer): pass
- server = StratumThreadedJSONRPCServer(( config.get('server','host'), 8081))
- server.register_function(get_peers_json, 'server.peers')
- server.register_function(cmd_stop, 'stop')
- server.register_function(cmd_load, 'load')
- server.register_function(get_banner, 'server.banner')
- server.register_function(lambda a,b,c: store.send_tx(c), 'transaction.broadcast')
- server.register_function(address_get_history_json, 'address.get_history')
- server.register_function(add_address_to_session_json, 'address.subscribe')
- server.register_function(subscribe_to_numblocks_json, 'numblocks.subscribe')
- server.register_function(client_version_json, 'client.version')
- server.register_function(create_session_json, 'session.create') # internal message (not part of protocol)
- server.register_function(poll_session_json, 'session.poll') # internal message (not part of protocol)
- server.serve_forever()
if __name__ == '__main__':
# backend
store = abe_backend.AbeStore(config)
- # supported protocols
+ # old protocol
thread.start_new_thread(native_server_thread, ())
-
- thread.start_new_thread(http_server_thread, ())
thread.start_new_thread(clean_session_thread, ())
- #tcp stratum
- stratum_processor = AbeProcessor()
- shared = stratum.Shared()
+ processor = AbeProcessor()
+ shared = Shared()
# Bind shared to processor since constructor is user defined
- stratum_processor.shared = shared
- stratum_processor.start()
+ processor.shared = shared
+ processor.start()
+ # dispatcher
+ dispatcher = Dispatcher(shared, processor)
+ dispatcher.start()
# Create various transports we need
- server = stratum.TcpServer(shared, stratum_processor, "ecdsa.org",50001)
- server.start()
+ transports = [ TcpServer(shared, processor, "ecdsa.org",50001),
+ HttpServer(shared, processor, "ecdsa.org",8081)
+ ]
+ for server in transports:
+ server.start()
- if (config.get('server','irc') == 'yes' ):
- thread.start_new_thread(irc_thread, ())
-
- print "starting Electrum server"
- old_block_number = None
- while not stopping:
- block_number = store.main_iteration()
-
- if block_number != old_block_number:
- old_block_number = block_number
- stratum_processor.update_from_blocknum(block_number)
-
- while True:
- try:
- addr = store.address_queue.get(False)
- except:
- break
+ if (config.get('server','irc') == 'yes' ):
+ irc = Irc(processor)
+ irc.start()
- stratum_processor.update_from_address(addr)
- time.sleep(10)
+ print "starting Electrum server"
+ store.run(processor)
print "server stopped"
git://git.novaco.in / electrum-server.git / blobdiff