Source.Python/addons/source-python/Python3/asynchat.py at master · durof/Source.Python · GitHub

Repository navigation

History

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

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

294

295

296

297

298

299

300

301

302

303

304

305

306

307

# -*- Mode: Python; tab-width: 4 -*-

# Id: asynchat.py,v 2.26 2000/09/07 22:29:26 rushing Exp

# Author: Sam Rushing <rushing@nightmare.com>

# ======================================================================

# Permission to use, copy, modify, and distribute this software and

# its documentation for any purpose and without fee is hereby

# granted, provided that the above copyright notice appear in all

# copies and that both that copyright notice and this permission

# notice appear in supporting documentation, and that the name of Sam

# Rushing not be used in advertising or publicity pertaining to

# distribution of the software without specific, written prior

# permission.

# SAM RUSHING DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE,

# INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS, IN

# NO EVENT SHALL SAM RUSHING BE LIABLE FOR ANY SPECIAL, INDIRECT OR

# CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS

# OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT,

# NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN

# CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.

# ======================================================================

r"""A class supporting chat-style (command/response) protocols.

This class adds support for 'chat' style protocols - where one side

sends a 'command', and the other sends a response (examples would be

the common internet protocols - smtp, nntp, ftp, etc..).

The handle_read() method looks at the input stream for the current

'terminator' (usually '\r\n' for single-line responses, '\r\n.\r\n'

for multi-line output), calling self.found_terminator() on its

receipt.

for example:

Say you build an async nntp client using this class. At the start

of the connection, you'll have self.terminator set to '\r\n', in

order to process the single-line greeting. Just before issuing a

'LIST' command you'll set it to '\r\n.\r\n'. The output of the LIST

command will be accumulated (using your own 'collect_incoming_data'

method) up to the terminator, and then control will be returned to

you - by calling your self.found_terminator() method.

"""

import asyncore

from collections import deque

class async_chat(asyncore.dispatcher):

"""This is an abstract class. You must derive from this class, and add

the two methods collect_incoming_data() and found_terminator()"""

# these are overridable defaults

ac_in_buffer_size = 65536

ac_out_buffer_size = 65536

# we don't want to enable the use of encoding by default, because that is a

# sign of an application bug that we don't want to pass silently

use_encoding = 0

encoding = 'latin-1'

def __init__(self, sock=None, map=None):

# for string terminator matching

self.ac_in_buffer = b''

# we use a list here rather than io.BytesIO for a few reasons...

# del lst[:] is faster than bio.truncate(0)

# lst = [] is faster than bio.truncate(0)

self.incoming = []

# we toss the use of the "simple producer" and replace it with

# a pure deque, which the original fifo was a wrapping of

self.producer_fifo = deque()

asyncore.dispatcher.__init__(self, sock, map)

def collect_incoming_data(self, data):

raise NotImplementedError("must be implemented in subclass")

def _collect_incoming_data(self, data):

self.incoming.append(data)

def _get_data(self):

d = b''.join(self.incoming)

del self.incoming[:]

return d

def found_terminator(self):

raise NotImplementedError("must be implemented in subclass")

def set_terminator(self, term):

"""Set the input delimiter.

Can be a fixed string of any length, an integer, or None.

"""

if isinstance(term, str) and self.use_encoding:

term = bytes(term, self.encoding)

elif isinstance(term, int) and term < 0:

raise ValueError('the number of received bytes must be positive')

self.terminator = term

def get_terminator(self):

return self.terminator

# grab some more data from the socket,

# throw it to the collector method,

# check for the terminator,

# if found, transition to the next state.

def handle_read(self):

try:

data = self.recv(self.ac_in_buffer_size)

except BlockingIOError:

return

except OSError as why:

self.handle_error()

return

if isinstance(data, str) and self.use_encoding:

data = bytes(str, self.encoding)

self.ac_in_buffer = self.ac_in_buffer + data

# Continue to search for self.terminator in self.ac_in_buffer,

# while calling self.collect_incoming_data. The while loop

# is necessary because we might read several data+terminator

# combos with a single recv(4096).

while self.ac_in_buffer:

lb = len(self.ac_in_buffer)

terminator = self.get_terminator()

if not terminator:

# no terminator, collect it all

self.collect_incoming_data(self.ac_in_buffer)

self.ac_in_buffer = b''

elif isinstance(terminator, int):

# numeric terminator

n = terminator

if lb < n:

self.collect_incoming_data(self.ac_in_buffer)

self.ac_in_buffer = b''

self.terminator = self.terminator - lb

else:

self.collect_incoming_data(self.ac_in_buffer[:n])

self.ac_in_buffer = self.ac_in_buffer[n:]

self.terminator = 0

self.found_terminator()

else:

# 3 cases:

# 1) end of buffer matches terminator exactly:

# collect data, transition

# 2) end of buffer matches some prefix:

# collect data to the prefix

# 3) end of buffer does not match any prefix:

# collect data

terminator_len = len(terminator)

index = self.ac_in_buffer.find(terminator)

if index != -1:

# we found the terminator

if index > 0:

# don't bother reporting the empty string

# (source of subtle bugs)

self.collect_incoming_data(self.ac_in_buffer[:index])

self.ac_in_buffer = self.ac_in_buffer[index+terminator_len:]

# This does the Right Thing if the terminator

# is changed here.

self.found_terminator()

else:

# check for a prefix of the terminator

index = find_prefix_at_end(self.ac_in_buffer, terminator)

if index:

if index != lb:

# we found a prefix, collect up to the prefix

self.collect_incoming_data(self.ac_in_buffer[:-index])

self.ac_in_buffer = self.ac_in_buffer[-index:]

break

else:

# no prefix, collect it all

self.collect_incoming_data(self.ac_in_buffer)

self.ac_in_buffer = b''

def handle_write(self):

self.initiate_send()

def handle_close(self):

self.close()

def push(self, data):

if not isinstance(data, (bytes, bytearray, memoryview)):

raise TypeError('data argument must be byte-ish (%r)',

type(data))

sabs = self.ac_out_buffer_size

if len(data) > sabs:

for i in range(0, len(data), sabs):

self.producer_fifo.append(data[i:i+sabs])

else:

self.producer_fifo.append(data)

self.initiate_send()

def push_with_producer(self, producer):

self.producer_fifo.append(producer)

self.initiate_send()

def readable(self):

"predicate for inclusion in the readable for select()"

# cannot use the old predicate, it violates the claim of the

# set_terminator method.

# return (len(self.ac_in_buffer) <= self.ac_in_buffer_size)

return 1

def writable(self):

"predicate for inclusion in the writable for select()"

return self.producer_fifo or (not self.connected)

def close_when_done(self):

"automatically close this channel once the outgoing queue is empty"

self.producer_fifo.append(None)

def initiate_send(self):

while self.producer_fifo and self.connected:

first = self.producer_fifo[0]

# handle empty string/buffer or None entry

if not first:

del self.producer_fifo[0]

if first is None:

self.handle_close()

return

# handle classic producer behavior

obs = self.ac_out_buffer_size

try:

data = first[:obs]

except TypeError:

data = first.more()

if data:

self.producer_fifo.appendleft(data)

else:

del self.producer_fifo[0]

continue

if isinstance(data, str) and self.use_encoding:

data = bytes(data, self.encoding)

# send the data

try:

num_sent = self.send(data)

except OSError:

self.handle_error()

return

if num_sent:

if num_sent < len(data) or obs < len(first):

self.producer_fifo[0] = first[num_sent:]

else:

del self.producer_fifo[0]

# we tried to send some actual data

return

def discard_buffers(self):

# Emergencies only!

self.ac_in_buffer = b''

del self.incoming[:]

self.producer_fifo.clear()

class simple_producer:

def __init__(self, data, buffer_size=512):

self.data = data

self.buffer_size = buffer_size

def more(self):

if len(self.data) > self.buffer_size:

result = self.data[:self.buffer_size]

self.data = self.data[self.buffer_size:]

return result

else:

result = self.data

self.data = b''

return result

# Given 'haystack', see if any prefix of 'needle' is at its end. This

# assumes an exact match has already been checked. Return the number of

# characters matched.

# for example:

# f_p_a_e("qwerty\r", "\r\n") => 1

# f_p_a_e("qwertydkjf", "\r\n") => 0

# f_p_a_e("qwerty\r\n", "\r\n") => <undefined>

# this could maybe be made faster with a computed regex?

# [answer: no; circa Python-2.0, Jan 2001]

# new python: 28961/s

# old python: 18307/s

# re: 12820/s

# regex: 14035/s

def find_prefix_at_end(haystack, needle):

l = len(needle) - 1

while l and not haystack.endswith(needle[:l]):

l -= 1

return l

Source.Python/addons/source-python/Python3/asynchat.py at master · durof/Source.Python · GitHub

Repository navigation

Expand file tree

Breadcrumbs

asynchat.py

Latest commit

History

Breadcrumbs

asynchat.py

File metadata and controls