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
|
--- a/txws.py
+++ b/txws.py
@@ -27,7 +27,6 @@
__version__ = "0.7.1"
-import six
import array
@@ -139,7 +138,7 @@
first = int("".join(i for i in key1 if i in digits)) // key1.count(" ")
second = int("".join(i for i in key2 if i in digits)) // key2.count(" ")
- nonce = pack(">II8s", first, second, six.b(challenge))
+ nonce = pack(">II8s", first, second, challenge.encode("latin-1"))
return md5(nonce).digest()
@@ -169,10 +168,10 @@
and valid text without any 0xff bytes.
"""
- if isinstance(buf, six.text_type):
+ if isinstance(buf, str):
buf = buf.encode('utf-8')
- return six.b("\x00") + buf + six.b("\xff")
+ return b"\x00" + buf + b"\xff"
def parse_hybi00_frames(buf):
"""
@@ -182,12 +181,12 @@
and will actively ignore it.
"""
- start = buf.find(six.b("\x00"))
+ start = buf.find(b"\x00")
tail = 0
frames = []
while start != -1:
- end = buf.find(six.b("\xff"), start + 1)
+ end = buf.find(b"\xff", start + 1)
if end == -1:
# Incomplete frame, try again later.
break
@@ -196,7 +195,7 @@
frame = buf[start + 1:end]
frames.append((NORMAL, frame))
tail = end + 1
- start = buf.find(six.b("\x00"), end + 1)
+ start = buf.find(b"\x00", end + 1)
# Adjust the buffer and return.
buf = buf[tail:]
@@ -231,12 +230,12 @@
else:
length = chr(len(buf))
- if isinstance(buf, six.text_type):
+ if isinstance(buf, str):
buf = buf.encode('utf-8')
# Always make a normal packet.
header = chr(0x80 | opcode)
- return six.b(header + length) + buf
+ return (header + length).encode("latin-1") + buf
def make_hybi07_frame_dwim(buf):
"""
@@ -244,9 +243,9 @@
"""
# TODO: eliminate magic numbers.
- if isinstance(buf, six.binary_type):
+ if isinstance(buf, bytes):
return make_hybi07_frame(buf, opcode=0x2)
- elif isinstance(buf, six.text_type):
+ elif isinstance(buf, str):
return make_hybi07_frame(buf.encode("utf-8"), opcode=0x1)
else:
raise TypeError("In binary support mode, frame data must be either str or unicode")
@@ -268,9 +267,6 @@
# about, and an opcode which nobody cares about.
header = buf[start]
- if six.PY2:
- header = ord(header)
-
if header & 0x70:
# At least one of the reserved flags is set. Pork chop sandwiches!
raise WSException("Reserved flag in HyBi-07 frame (%d)" % header)
@@ -289,9 +285,6 @@
# extra length.
length = buf[start + 1]
- if six.PY2:
- length = ord(length)
-
masked = length & 0x80
length &= 0x7f
@@ -342,7 +335,7 @@
data = unpack(">H", data[:2])[0], data[2:]
else:
# No reason given; use generic data.
- data = 1000, six.b("No reason given")
+ data = 1000, b"No reason given"
frames.append((opcode, data))
start += offset + length
@@ -355,7 +348,7 @@
layer.
"""
- buf = six.b("")
+ buf = b""
codec = None
location = "/"
host = "example.com"
@@ -385,7 +378,7 @@
return ISSLTransport(self.transport, None) is not None
def writeEncoded(self, data):
- if isinstance(data, six.text_type):
+ if isinstance(data, str):
data = data.encode('utf-8')
self.transport.write(data)
@@ -528,7 +521,7 @@
elif "Sec-WebSocket-Protocol" in self.headers:
protocols = self.headers["Sec-WebSocket-Protocol"]
- if isinstance(protocols, six.string_types):
+ if isinstance(protocols, str):
protocols = [p.strip() for p in protocols.split(',')]
for protocol in protocols:
@@ -587,7 +580,7 @@
# These lines look like:
# GET /some/path/to/a/websocket/resource HTTP/1.1
if self.state == REQUEST:
- separator = six.b("\r\n")
+ separator = b"\r\n"
if separator in self.buf:
request, chaff, self.buf = self.buf.partition(separator)
request = request.decode('utf-8')
@@ -601,7 +594,7 @@
elif self.state == NEGOTIATING:
# Check to see if we've got a complete set of headers yet.
- separator = six.b("\r\n\r\n")
+ separator = b"\r\n\r\n"
if separator in self.buf:
head, chaff, self.buf = self.buf.partition(separator)
head = head.decode('utf-8')
|
