/usr/lib/python2.7/dist-packages/fs/remotefs.py is in python-fs 0.5.4-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 | # Work in Progress - Do not use
from __future__ import with_statement
from fs.base import FS
from fs.expose.serve import packetstream
from collections import defaultdict
import threading
from threading import Lock, RLock
from json import dumps
import Queue as queue
import socket
from six import b
class PacketHandler(threading.Thread):
def __init__(self, transport, prelude_callback=None):
super(PacketHandler, self).__init__()
self.transport = transport
self.encoder = packetstream.JSONFileEncoder(transport)
self.decoder = packetstream.JSONDecoder(prelude_callback=None)
self.queues = defaultdict(queue.Queue)
self._encoder_lock = threading.Lock()
self._queues_lock = threading.Lock()
self._call_id_lock = threading.Lock()
self.call_id = 0
def run(self):
decoder = self.decoder
read = self.transport.read
on_packet = self.on_packet
while True:
data = read(1024*16)
if not data:
print "No data"
break
print "data", repr(data)
for header, payload in decoder.feed(data):
print repr(header)
print repr(payload)
on_packet(header, payload)
def _new_call_id(self):
with self._call_id_lock:
self.call_id += 1
return self.call_id
def get_thread_queue(self, queue_id=None):
if queue_id is None:
queue_id = threading.current_thread().ident
with self._queues_lock:
return self.queues[queue_id]
def send_packet(self, header, payload=''):
call_id = self._new_call_id()
queue_id = threading.current_thread().ident
client_ref = "%i:%i" % (queue_id, call_id)
header['client_ref'] = client_ref
with self._encoder_lock:
self.encoder.write(header, payload)
return call_id
def get_packet(self, call_id):
if call_id is not None:
queue_id = threading.current_thread().ident
client_ref = "%i:%i" % (queue_id, call_id)
else:
client_ref = None
queue = self.get_thread_queue()
while True:
header, payload = queue.get()
print repr(header)
print repr(payload)
if client_ref is not None and header.get('client_ref') != client_ref:
continue
break
return header, payload
def on_packet(self, header, payload):
client_ref = header.get('client_ref', '')
queue_id, call_id = client_ref.split(':', 1)
queue_id = int(queue_id)
#queue_id = header.get('queue_id', '')
queue = self.get_thread_queue(queue_id)
queue.put((header, payload))
class _SocketFile(object):
def __init__(self, socket):
self.socket = socket
def read(self, size):
try:
return self.socket.recv(size)
except:
return b('')
def write(self, data):
self.socket.sendall(data)
def close(self):
self.socket.shutdown(socket.SHUT_RDWR)
self.socket.close()
class _RemoteFile(object):
def __init__(self, path, connection):
self.path = path
self.connection = connection
class RemoteFS(FS):
_meta = { 'thead_safe' : True,
'network' : True,
'virtual' : False,
'read_only' : False,
'unicode_paths' : True,
}
def __init__(self, addr='', port=3000, username=None, password=None, resource=None, transport=None):
self.addr = addr
self.port = port
self.username = None
self.password = None
self.resource = None
self.transport = transport
if self.transport is None:
self.transport = self._open_connection()
self.packet_handler = PacketHandler(self.transport)
self.packet_handler.start()
self._remote_call('auth',
username=username,
password=password,
resource=resource)
def _open_connection(self):
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect((self.addr, self.port))
socket_file = _SocketFile(sock)
socket_file.write(b('pyfs/0.1\n'))
return socket_file
def _make_call(self, method_name, *args, **kwargs):
call = dict(type='rpc',
method=method_name,
args=args,
kwargs=kwargs)
return call
def _remote_call(self, method_name, *args, **kwargs):
call = self._make_call(method_name, *args, **kwargs)
call_id = self.packet_handler.send_packet(call)
header, payload = self.packet_handler.get_packet(call_id)
return header, payload
def ping(self, msg):
call_id = self.packet_handler.send_packet({'type':'rpc', 'method':'ping'}, msg)
header, payload = self.packet_handler.get_packet(call_id)
print "PING"
print header
print payload
def close(self):
self.transport.close()
self.packet_handler.join()
def open(self, path, mode="r", **kwargs):
pass
def exists(self, path):
remote = self._remote_call('exists', path)
return remote.get('response')
if __name__ == "__main__":
rfs = RemoteFS()
rfs.close()
|