FazBrowse GitHub Viewer
|
Trending
|
URL:
|
Home
Tools:
[Download Repo ZIP]
[View Raw Code]
[Original HTTPS Page]
python-client/ldclient/streaming.py at master · dronedeploy/python-client · GitHub
Uh oh!
There was an error while loading.
Please reload this page
.
This repository was archived by the owner on May 5, 2022. It is now read-only.
dronedeploy
/
python-client
Public archive
forked from
launchdarkly/python-server-sdk
Notifications
You must be signed in to change notification settings
Fork
0
Star
0
Code
Pull requests
0
Actions
Wiki
Security and quality
0
Insights
Additional navigation options
Code
Pull requests
Actions
Wiki
Security and quality
Insights
Expand file tree
Breadcrumbs
python-client
/
ldclient
/
streaming.py
Copy path
More file actions
More file actions
Latest commit
History
History
History
75 lines (66 loc) · 2.75 KB
Breadcrumbs
python-client
/
ldclient
/
streaming.py
Copy path
File metadata and controls
75 lines (66 loc) · 2.75 KB
Raw
Copy raw file
Download raw file
Open symbols panel
Edit and raw actions
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
import
json
from
threading
import
Thread
import
time
from
sseclient
import
SSEClient
from
ldclient
.
interfaces
import
UpdateProcessor
from
ldclient
.
util
import
_stream_headers
,
log
class
StreamingUpdateProcessor
(
Thread
,
UpdateProcessor
):
def
__init__
(
self
,
sdk_key
,
config
,
requester
,
store
,
ready
):
Thread
.
__init__
(
self
)
self
.
daemon
=
True
self
.
_sdk_key
=
sdk_key
self
.
_config
=
config
self
.
_requester
=
requester
self
.
_store
=
store
self
.
_running
=
False
self
.
_ready
=
ready
def
run
(
self
):
log
.
info
(
"Starting StreamingUpdateProcessor connecting to uri: "
+
self
.
_config
.
stream_uri
)
self
.
_running
=
True
hdrs
=
_stream_headers
(
self
.
_sdk_key
)
uri
=
self
.
_config
.
stream_uri
while
self
.
_running
:
try
:
messages
=
SSEClient
(
uri
,
verify
=
self
.
_config
.
verify_ssl
,
headers
=
hdrs
)
for
msg
in
messages
:
if
not
self
.
_running
:
break
self
.
process_message
(
self
.
_store
,
self
.
_requester
,
msg
,
self
.
_ready
)
except
Exception
as
e
:
log
.
error
(
"Could not connect to LaunchDarkly stream: "
+
str
(
e
.
message
)
+
" waiting 1 second before trying again."
)
time
.
sleep
(
1
)
def
stop
(
self
):
log
.
info
(
"Stopping StreamingUpdateProcessor"
)
self
.
_running
=
False
def
initialized
(
self
):
return
self
.
_running
and
self
.
_ready
.
is_set
()
and
self
.
_store
.
initialized
@
staticmethod
def
process_message
(
store
,
requester
,
msg
,
ready
):
payload
=
json
.
loads
(
msg
.
data
)
log
.
debug
(
"Received stream event {}"
.
format
(
msg
.
event
))
if
msg
.
event
==
'put'
:
store
.
init
(
payload
)
if
not
ready
.
is_set
()
and
store
.
initialized
:
ready
.
set
()
log
.
info
(
"StreamingUpdateProcessor initialized ok"
)
elif
msg
.
event
==
'patch'
:
key
=
payload
[
'path'
][
1
:]
feature
=
payload
[
'data'
]
log
.
debug
(
"Updating feature {}"
.
format
(
key
))
store
.
upsert
(
key
,
feature
)
elif
msg
.
event
==
"indirect/patch"
:
key
=
payload
[
'data'
]
store
.
upsert
(
key
,
requester
.
get_one
(
key
))
elif
msg
.
event
==
"indirect/put"
:
store
.
init
(
requester
.
get_all
())
if
not
ready
.
is_set
()
and
store
.
initialized
:
ready
.
set
()
log
.
info
(
"StreamingUpdateProcessor initialized ok"
)
elif
msg
.
event
==
'delete'
:
key
=
payload
[
'path'
][
1
:]
# noinspection PyShadowingNames
version
=
payload
[
'version'
]
store
.
delete
(
key
,
version
)
else
:
log
.
warning
(
'Unhandled event in stream processor: '
+
msg
.
event
)
Back
|
FazBrowse Home
|
New Git URL