-
Notifications
You must be signed in to change notification settings - Fork 0
/
PipeW.py
93 lines (82 loc) · 2.43 KB
/
PipeW.py
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
import os
import time
import io
import threading
import queue
from threading import Thread, Lock
class PipeW:
def __init__(self, fifoPath, msgQMax):
self.path = fifoPath
self.pipe = None
self.blocking = False
self.msgQ = queue.Queue()
self.msgQMax = msgQMax
self.pipeThread = Thread(target = self.PipeThreadHandler)
def OpenPipe(self):
try:
os.mkfifo(self.path)
print("Created pipe on filesystem: " . self.path)
except Exception as e:
print("MKFIFO: " + repr(e))
try:
print("Waiting for listener.")
self.blocking = True
fd = os.open(self.path, os.O_WRONLY)# blocking
self.blocking = False
print("OPENED File")
out = os.fdopen(fd, 'w') #also non-blocking
print("OPENED File descriptor")
print(io.DEFAULT_BUFFER_SIZE)
except Exception as e:
out = None
print("OPEN: " . e)
return out
def ClosePipe(self):
try:
if self.blocking == False:
self.pipe.flush()
self.pipe.close()
else:
self.pipeThreadRunning = False
except BrokenPipeError:
print("Pipe was already closed.")
self.pipe = None
try:
os.unlink(self.path)
except FileNotFoundError:
pass
def PipeThreadHandler(self):
self.pipeThreadRunning = True
self.pipe = self.OpenPipe()
while self.pipeThreadRunning:
try:
msg = self.msgQ.get(block = True, timeout = 2)
self.msgQ.task_done()
self.pipe.write(msg)
self.pipe.flush()
except queue.Empty:
pass
except BrokenPipeError:
print("Disconnected!")
self.ClosePipe()
self.pipe = self.OpenPipe()
self.ClosePipe()
def Send(self, message):
if self.msgQ.qsize() < self.msgQMax:
self.msgQ.put(message)
else:
self.msgQ.get(block = False)
self.msgQ.task_done()
self.msgQ.put(message)
def Start(self):
if self.pipeThread.is_alive() == False:
print("Starting pipe thread.")
self.pipeThread.start()
def Stop(self):
if self.pipeThread.is_alive() == True:
self.pipeThreadRunning = False
if self.blocking == True:
fifo = os.open(self.path, os.O_RDONLY)
os.close(fifo)
else:
self.ClosePipe()