Skip to content

proxy: Add detection for stale/broken TCP connections #724

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
hlchau wants to merge 1 commit into
base: master
Choose a base branch
from
Open

proxy: Add detection for stale/broken TCP connections #724

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions grottconf.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,9 @@ def __init__(self, vrm):
self.grottip = "default" #connect to server IP adress
self.outfile ="sys.stdout"
self.tmzone = "local" #set timezone (at this moment only used for influxdb)

self.select_timeout_s = 60 * 1 # Timeout for select()
self.client_timeout_s = 60 * 5 # Time without any received packets from growatt device until it will be considered as disconnected

#Growatt server default
#self.growattip = "47.91.67.66"
Expand Down
31 changes: 30 additions & 1 deletion grottproxy.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,7 @@ def start(self, host, port):
return False

class Proxy:
client_timeouts_s = {}
input_list = []
channel = {}

Expand Down Expand Up @@ -126,13 +127,32 @@ def main(self,conf):
while 1:
time.sleep(delay)
ss = select.select
inputready, outputready, exceptready = ss(self.input_list, [], [])
inputready, outputready, exceptready = ss(self.input_list, [], [], conf.select_timeout_s)
if (not inputready):
for key in list(self.client_timeouts_s.keys()):
# Skip sockets, which have been removed in the meantime.
if (key not in self.client_timeouts_s):
continue
# Increase timeout counter.
self.client_timeouts_s[key] += conf.select_timeout_s
# If the timeout limit was reached, consider the socket to be broken or already
# closed by the remote.
if (self.client_timeouts_s[key] >= conf.client_timeout_s):
try:
print("\t - Grott connection closed (timeout): remote={}".format(key.getpeername()))
except:
print("\t - Grott connection closed (timeout)")
self.s = key
self.on_close(conf)
continue
for self.s in inputready:
if self.s == self.server:
self.on_accept(conf)
break
try:
self.data, self.addr = self.s.recvfrom(buffer_size)
# Reset timeout counter for the socket we just received from.
self.client_timeouts_s[self.s] = 0
except:
if conf.verbose : print("\t - Grott connection error")
self.on_close(conf)
Expand All @@ -152,6 +172,9 @@ def on_accept(self,conf):
self.input_list.append(forward)
self.channel[clientsock] = forward
self.channel[forward] = clientsock
# Initialize timeout counters for the new proxy and forwarder socket.
self.client_timeouts_s[clientsock] = 0
self.client_timeouts_s[forward] = 0
else:
if conf.verbose:
print("\t - Can't establish connection with remote server."),
Expand All @@ -177,6 +200,10 @@ def on_close(self,conf):
# delete both objects from channel dict
del self.channel[out]
del self.channel[self.s]
# Delete timeout counter for proxy/grott socket.
del self.client_timeouts_s[self.s]
# Delete timeout counter for forwarder/growatt socket.
del self.client_timeouts_s[out]

def on_recv(self,conf):
data = self.data
Expand Down Expand Up @@ -233,6 +260,8 @@ def on_recv(self,conf):

# send data to destination
self.channel[self.s].send(data)
# Reset timeout counter for the socket we just sent data to.
self.client_timeouts_s[self.channel[self.s]] = 0
if len(data) > conf.minrecl :
#process received data
procdata(conf,data)
Expand Down