1 """Management for daemon processes
3 This module provides some client support for the daemon management
4 provided in the pdm.srv module.
7 import socket, pickle, struct, select, threading
9 __all__ = ["client", "replclient"]
11 class protoerr(Exception):
15 if isinstance(spec, socket.socket):
20 sk = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
23 sk = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
24 sk.connect(("localhost", int(spec)))
26 sk = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
28 sk.connect((spec[:p], int(spec[p + 1:])))
30 raise Exception("Unknown target specification %r" % spec)
34 if sk is not None: sk.close()
38 def __init__(self, sk, proto = None):
41 line = self.readline()
43 raise protoerr("Illegal protocol signature")
52 p = self.buf.find(b"\n")
55 self.buf = self.buf[p + 1:]
57 ret = self.sk.recv(1024)
62 def select(self, proto):
63 if isinstance(proto, str):
64 proto = proto.encode("ascii")
66 raise Exception("Illegal protocol specified: %r" % proto)
67 self.sk.send(proto + b"\n")
69 if len(rep) < 1 or rep[0] != b"+"[0]:
70 raise protoerr("Error reply when selecting protocol %s: %s" % (proto, rep[1:]))
75 def __exit__(self, *excinfo):
79 class replclient(client):
80 def __init__(self, sk):
81 super(replclient, self).__init__(sk, "repl")
85 ncode = code.replace("\n\n", "\n")
86 if ncode == code: break
88 while len(code) > 0 and code[-1] == "\n":
90 self.sk.send((code + "\n\n").encode("utf-8"))
96 elif ln[0] == b"+"[0]:
97 return buf.decode("utf-8")
98 elif ln[0] == b"-"[0]:
99 raise protoerr("Error reply: %s" % ln[1:].decode("utf-8"))
101 raise protoerr("Illegal reply: %s" % ln)
103 class perfproxy(object):
104 def __init__(self, cl, id, proto):
108 self.subscribers = set()
110 def lookup(self, name):
111 self.cl.lock.acquire()
116 self.cl.lock.release()
117 (proto,) = self.cl.run("lookup", id, self.id, name)
118 proxy = perfproxy(self.cl, id, proto)
119 self.cl.proxies[id] = proxy
123 return self.cl.run("ls", self.id)[0]
126 return self.cl.run("readattr", self.id)[0]
129 return self.cl.run("attrinfo", self.id)[0]
131 def invoke(self, method, *args, **kwargs):
132 return self.cl.run("invoke", self.id, method, args, kwargs)[0]
134 def subscribe(self, cb):
135 if cb in self.subscribers:
136 raise ValueError("Already subscribed")
137 if len(self.subscribers) == 0:
138 self.cl.run("subs", self.id)
139 self.subscribers.add(cb)
141 def unsubscribe(self, cb):
142 if cb not in self.subscribers:
143 raise ValueError("Not subscribed")
144 self.subscribers.remove(cb)
145 if len(self.subscribers) == 0:
146 self.cl.run("unsubs", self.id)
148 def notify(self, ev):
149 for cb in self.subscribers:
155 self.cl.run("unbind", self.id)
156 del self.cl.proxies[self.id]
161 def __exit__(self, *excinfo):
165 class perfclient(client):
166 def __init__(self, sk):
167 super(perfclient, self).__init__(sk, "perf")
169 self.lock = threading.Lock()
174 buf = pickle.dumps(ob)
175 buf = struct.pack(">l", len(buf)) + buf
178 def recvb(self, num):
180 while len(buf) < num:
181 data = self.sk.recv(num - len(buf))
188 return pickle.loads(self.recvb(struct.unpack(">l", self.recvb(4))[0]))
190 def event(self, id, ev):
191 proxy = self.proxies.get(id)
192 if proxy is None: return
195 def dispatch(self, timeout = None):
196 rfd, wfd, efd = select.select([self.sk], [], [], timeout)
200 self.event(msg[1], msg[2])
202 raise ValueError("Unexpected non-event message: %r" % msg[0])
207 if reply[0] in ("+", "-"):
209 elif reply[0] == "*":
210 self.event(reply[1], reply[2])
212 raise ValueError("Illegal reply header: %r" % reply[0])
214 def run(self, cmd, *args):
217 self.send((cmd,) + args)
218 reply = self.recvreply()
226 def lookup(self, module, obnm):
233 (proto,) = self.run("bind", id, module, obnm)
234 proxy = perfproxy(self, id, proto)
235 self.proxies[id] = proxy
238 def find(self, name):
239 ret = self.names.get(name)
243 ret = self.find(name[:p]).lookup(name[p + 1:])
246 ret = self.lookup(name[:p], name[p + 1:])
247 self.names[name] = ret