/[de-setf-amqp]/frames.lisp
ViewVC logotype

Contents of /frames.lisp

Parent Directory Parent Directory | Revision Log Revision Log


Revision 3 - (show annotations)
Tue Feb 23 09:05:39 2010 UTC (4 years, 1 month ago) by janderson
File size: 29725 byte(s)
Merge commit 'remotes/github/master' into remotes/git-svn
1 ;;; -*- Package: de.setf.amqp.implementation; -*-
2
3 (in-package :de.setf.amqp.implementation)
4
5 (document :file
6 (description "This file defines the wire-level frame model for for the `de.setf.amqp` Connon Lisp library.")
7 (copyright
8 "Copyright 2010 [james anderson](mailto:james.anderson@setf.de) All Rights Reserved"
9 "'de.setf.amqp' is free software: you can redistribute it and/or modify it under the terms of version 3
10 of the GNU Affero General Public License as published by the Free Software Foundation.
11
12 'setf.amqp' is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the
13 implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
14 See the Affero General Public License for more details.
15
16 A copy of the GNU Affero General Public License should be included with 'de.setf.amqp' as `AMQP:agpl.txt`.
17 If not, see the GNU [site](http://www.gnu.org/licenses/).")
18
19 (long-description "Each protocol version packages the data on the wire in a slightly different form. The
20 `frame` is specialized for each version to implement operators which marshall/unmarshall frame data as
21 as per the version, and operators which map between abstract objcet and method class names and the
22 version-specific numeric identifiers."))
23
24
25 (defclass amqp:frame ()
26 ((connection
27 :initarg :connection :initform nil ;(error "connection required.")
28 :reader frame-connection)
29 (header
30 :initarg :header :initform nil ;(error "header required")
31 :reader frame-header)
32 (data
33 :initarg :buffer :initarg :data :initform nil ; (error "buffer required.")
34 :reader frame-data :writer setf-frame-data :writer (setf frame-data)
35 :type (simple-array (unsigned-byte 8) (*))
36 :documentation "A byte buffer serves to capture the i/o for any frames
37 which are not streamed. the encoding operators use the buffer if it is present.")
38 (end-marker :initform #xce :allocation :class
39 :reader frame-end-marker))
40 (:documentation "Comprises the header and payload for a single AMQP frame.
41 Specializations are defined for input and output and for the respective
42 protocol version. The former distinguish methods to treat the data as
43 received or sent while the latter distinguish header format."))
44
45
46 (defmethod amqp:ensure-method ((class amqp:object) (frame amqp:frame) &rest initargs)
47 (declare (dynamic-extent initargs))
48 (apply #'amqp:ensure-method class (frame-method-code frame) initargs))
49
50 (defmethod amqp:ensure-object ((context t) (frame amqp:frame) &rest initargs)
51 (declare (dynamic-extent initargs))
52 (apply #'amqp:ensure-object context (frame-class-name frame) initargs))
53
54
55 (defclass output-frame (amqp:frame)
56 ())
57
58 (defclass input-frame (amqp:frame)
59 ())
60
61 (defclass 7-byte-header-frame (amqp:frame)
62 ((input-buffer :type (simple-array (unsigned-byte 8) 7))))
63
64 (defclass amqp.u:7-byte-header-input-frame (7-byte-header-frame input-frame) ())
65 (defclass amqp.u:7-byte-header-output-frame (7-byte-header-frame output-frame) ())
66
67
68 (defclass 8-byte-header-frame (amqp:frame)
69 ((input-buffer :type (simple-array (unsigned-byte 8) 8))))
70
71 (defclass amqp.u:8-byte-header-input-frame (8-byte-header-frame input-frame) ())
72 (defclass amqp.u:8-byte-header-output-frame (8-byte-header-frame output-frame) ())
73
74
75 (defclass 12-byte-header-frame (amqp:frame)
76 ((input-buffer :type (simple-array (unsigned-byte 8) 12))))
77
78 (defclass amqp.u:12-byte-header-input-frame (12-byte-header-frame input-frame) ())
79 (defclass amqp.u:12-byte-header-output-frame (12-byte-header-frame output-frame) ())
80
81
82
83 (defgeneric format-frame (frame stream)
84 (:method ((frame amqp::frame) stream)
85 (let ((header (when (slot-boundp frame 'header) (frame-header frame)))
86 (data (when (slot-boundp frame 'data) (frame-data frame)))
87 (connection (when (slot-boundp frame 'connection) (frame-connection frame)))
88 (class-name nil))
89 (when header
90 (format stream "[(~:[-~;+~])~:[?~;~d|~d|c.~d|t.~d~]]"
91 connection
92 header
93 (setf class-name (frame-type-class-name frame))
94 (frame-cycle frame)
95 (frame-channel-number frame)
96 (frame-track-number frame)))
97 (write-string "." stream)
98 (when (eq class-name 'amqp:method)
99 (format stream "{~a.~a}"
100 (frame-class-name frame)
101 (frame-method-name frame)))
102 (format stream "[~:[?~;~@[~a~]/~d~]:"
103 data
104 (when header (frame-size frame))
105 (length data))
106 (let* ((size (if header (min (frame-size frame) (length data)) (length data)))
107 (beginning (min size 8))
108 (end (min 8 (- size beginning))))
109 (dotimes (x beginning) (format stream " ~0,'2d" (aref data x)))
110 (when (< (+ beginning end) size)
111 (write-string " ..." stream))
112 (dotimes (x end) (format stream " ~0,'2d" (aref data (+ x (- size end))))))
113 (write-string "]" stream))))
114
115 (defgeneric format-frame-header (frame stream)
116 (:method ((frame amqp::frame) stream)
117 (let ((header (when (slot-boundp frame 'header) (frame-header frame)))
118 (connection (when (slot-boundp frame 'connection) (frame-connection frame)))
119 (class-name nil))
120 (when header
121 (format stream "[(~:[-~;+~])~:[?~;~d|~d|c.~d|t.~d~]]"
122 connection
123 header
124 (setf class-name (frame-type-class-name frame))
125 (frame-cycle frame)
126 (frame-channel-number frame)
127 (frame-track-number frame)))
128 (write-string "." stream)
129 (when (eq class-name 'amqp:method)
130 (format stream "{~a.~a}"
131 (frame-class-name frame)
132 (frame-method-name frame)))
133 (write-string "]" stream))))
134
135
136 (defmethod print-object ((object amqp::frame) (stream t))
137 (print-unreadable-object (object stream :type t :identity t)
138 (format-frame object stream)))
139
140
141 ;;;
142 ;;;
143
144 (defgeneric release-frame (frame)
145 (:documentation "Release the frame.
146 This is specialized for input, respective output frames to return the
147 frame back to the respective free pool.")
148
149 (:method ((frame input-frame))
150 (when (frame-connection frame)
151 (setf-frame-channel-number 0 frame))
152 (enqueue frame (device-free-input-frames (frame-connection frame))))
153
154 (:method ((frame output-frame))
155 (when (frame-connection frame)
156 (setf-frame-channel-number 0 frame)
157 (enqueue frame (device-free-output-frames (frame-connection frame))))))
158
159
160 (defmacro with-input-frame ((frame device) &body body)
161 `(let* ((,frame (claim-input-frame ,device)))
162 (unwind-protect (progn ,@body)
163 (release-frame ,frame))))
164
165
166 (defgeneric make-input-frame (connection &rest args)
167 (:documentation "Allocate a new output frame and bind it to the
168 connection. The concrete frame class will depend on the connection version.
169 Given a channel, delegate to the respective connection.")
170 (declare (dynamic-extent args))
171
172 (:method ((channel amqp:channel) &rest args)
173 (declare (dynamic-extent args))
174 (apply #'make-input-frame (channel-connection channel) args))
175
176 (:method ((connection amqp:connection) &rest args)
177 (declare (dynamic-extent args))
178 (apply #'make-instance (connection-input-frame-class connection)
179 :connection connection
180 args)))
181
182
183 (defgeneric make-output-frame (connection &rest args)
184 (:documentation "Allocate a new input frame and bind it to the
185 connection. The concrete frame class will depend on the connection version.
186 Given a channel, delegate to the respective connection.")
187 (declare (dynamic-extent args))
188
189 (:method ((channel amqp:channel) &rest args)
190 (declare (dynamic-extent args))
191 (apply #'make-output-frame (channel-connection channel) args))
192
193 (:method ((connection amqp:connection) &rest args)
194 (declare (dynamic-extent args))
195 (apply #'make-instance (connection-output-frame-class connection)
196 :connection connection
197 args)))
198
199
200 (defgeneric put-encoded-frame (connection frame)
201 (:documentation "Place an encoded frame in the output queue. If the queue is empty.
202 write through. If the frame is written, release it.")
203
204 (:method ((channel amqp:channel) (frame output-frame))
205 (put-encoded-frame (channel-connection channel) frame))
206
207 (:method ((connection amqp:connection) (frame output-frame))
208 (flet ((write-encoded-frames ()
209 (do ((frame (dequeue (device-encoded-frames connection) :if-empty nil)
210 (dequeue (device-encoded-frames connection) :if-empty nil)))
211 ((or (null frame) (not (open-stream-p connection))))
212 (write-frame connection frame)
213 (release-frame frame))))
214 (declare (dynamic-extent #'write-encoded-frames))
215 (enqueue frame (device-encoded-frames connection)
216 :if-empty #'write-encoded-frames))))
217
218
219 (defgeneric get-encoded-frame (connection )
220 (:documentation "Get an encoded frame from the output queue. If the queue is empty.
221 return nil.")
222
223 (:method ((connection amqp:connection) )
224 (dequeue (device-encoded-frames connection) :if-empty nil)))
225
226
227 (defgeneric put-read-frame (connection frame)
228 (:documentation "Add the frame to the read frame queue.")
229
230 (:method ((connection amqp:connection) (frame input-frame))
231 (enqueue frame (device-read-frames connection)))
232
233 (:method ((connection amqp:channel) (frame input-frame))
234 (enqueue frame (device-read-frames connection))))
235
236
237 (defgeneric get-read-frame (connection &key wait)
238 (:documentation "Return the next read input frame for the given context.
239 A connection dequeues or reads. A channel dequeues or delegates to the
240 connection.")
241
242 (:method ((channel amqp:channel) &key (wait t))
243 (labels ((read-channel-frame ()
244 (let ((connection (channel-connection channel)))
245 (loop (if (open-stream-p connection)
246 (let ((frame (read-frame connection (claim-input-frame channel))))
247 (if (frame-matches-channel-p frame)
248 (return frame)
249 (put-read-frame channel frame))) ; this is the same as the connection queue
250 (return)))))
251 (frame-matches-channel-p (frame)
252 (eql (frame-channel-number frame) (channel-number channel))))
253 (declare (dynamic-extent #'frame-matches-channel-p #'read-channel-frame))
254 (dequeue (device-read-frames channel)
255 :test #'frame-matches-channel-p
256 :if-empty (when (or wait (stream-listen channel))
257 #'read-channel-frame))))
258
259 (:method ((connection amqp:connection) &key (wait t))
260 (labels ((read-connection-frame ()
261 (loop (let ((frame (read-frame connection (claim-input-frame connection))))
262 (if (frame-matches-connection-p frame)
263 (return frame)
264 (put-read-frame connection frame)))))
265 (frame-matches-connection-p (frame)
266 (eql (frame-channel-number frame) 0)))
267 (declare (dynamic-extent #'frame-matches-connection-p #'read-connection-frame))
268 (dequeue (device-read-frames connection)
269 :test #'frame-matches-connection-p
270 :if-empty (when (or wait (stream-listen connection))
271 #'read-connection-frame)))))
272 ;;;
273 ;;; frame accessors
274
275 (defgeneric frame-overhead (frame)
276 (:method ((frame 7-byte-header-frame))
277 8)
278
279 (:method ((frame 8-byte-header-frame))
280 9)
281
282 (:method ((frame 12-byte-header-frame))
283 12))
284
285
286 (defgeneric frame-decoder (frame)
287 (:method ((frame amqp:frame))
288 (c2mop:class-prototype (find-class (frame-type-class-name frame)))))
289
290 (defgeneric frame-type (frame)
291 (:method ((frame 7-byte-header-frame))
292 (buffer-unsigned-byte-8 (frame-header frame) 0))
293
294 (:method ((frame 8-byte-header-frame))
295 (buffer-unsigned-byte-8 (frame-header frame) 0))
296
297 (:method ((frame 12-byte-header-frame))
298 (buffer-unsigned-byte-8 (frame-header frame) 1)))
299
300 (defgeneric (setf frame-type) (type frame)
301 (:method (type (frame 7-byte-header-frame))
302 (setf (buffer-unsigned-byte-8 (frame-header frame) 0) type))
303
304 (:method (type (frame 8-byte-header-frame))
305 (setf (buffer-unsigned-byte-8 (frame-header frame) 0) type))
306
307 (:method (type (frame 12-byte-header-frame))
308 (setf (buffer-unsigned-byte-8 (frame-header frame) 1) type)))
309
310
311 (defgeneric frame-type-class-name (frame)
312 (:method ((frame amqp:frame))
313 (connection-frame-type-class-name (frame-connection frame) (frame-type frame))))
314
315 (defgeneric connection-frame-type-class-name (connection type-code))
316
317
318 (defgeneric setf-frame-type-class-name (name frame)
319 (:method ((name t) (frame amqp:frame))
320 (setf (frame-type frame) (connection-class-name-frame-type (frame-connection frame) name))
321 name))
322
323 (defgeneric connection-class-name-frame-type (connection type-code))
324
325 (setf (fdefinition '(setf frame-type-class-name))
326 #'setf-frame-type-class-name)
327
328
329 (defgeneric frame-cycle (frame)
330 (:method ((frame 7-byte-header-frame))
331 0)
332
333 (:method ((frame 8-byte-header-frame))
334 (buffer-unsigned-byte-8 (frame-header frame) 1))
335
336 (:method ((frame 12-byte-header-frame))
337 0))
338
339
340 (defgeneric setf-frame-cycle (cycle frame)
341 (:method (cycle (frame 7-byte-header-frame))
342 cycle)
343
344 (:method (cycle (frame 8-byte-header-frame))
345 (setf (buffer-unsigned-byte-8 (frame-header frame) 1) cycle))
346
347 (:method (cycle (frame 12-byte-header-frame))
348 cycle))
349
350
351 (defgeneric frame-channel-number (frame)
352 (:method ((frame 7-byte-header-frame))
353 (buffer-unsigned-byte-16 (frame-header frame) 1))
354
355 (:method ((frame 8-byte-header-frame))
356 (buffer-unsigned-byte-16 (frame-header frame) 2))
357
358 (:method ((frame 12-byte-header-frame))
359 (buffer-unsigned-byte-16 (frame-header frame) 6)))
360
361 (defgeneric setf-frame-channel-number (number frame)
362 (:method (number (frame 7-byte-header-frame))
363 (setf (buffer-unsigned-byte-16 (frame-header frame) 1) number))
364
365 (:method (number (frame 8-byte-header-frame))
366 (setf (buffer-unsigned-byte-16 (frame-header frame) 2) number))
367
368 (:method (number (frame 12-byte-header-frame))
369 number
370 (error "NYI")))
371
372 (setf (fdefinition '(setf frame-channel-number))
373 #'setf-frame-channel-number)
374
375
376 (defgeneric frame-track-number (frame)
377 (:method ((frame 7-byte-header-frame))
378 0)
379
380 (:method ((frame 8-byte-header-frame))
381 0)
382
383 (:method ((frame 12-byte-header-frame))
384 (logand (buffer-unsigned-byte-8 (frame-header frame) 5) #x0f)))
385
386 (defgeneric setf-frame-track-number (number frame)
387 (:method (number (frame 7-byte-header-frame))
388 number)
389
390 (:method (number (frame 8-byte-header-frame))
391 number)
392
393 (:method (number (frame 12-byte-header-frame))
394 (setf (buffer-unsigned-byte-8 (frame-header frame) 5)
395 (logior (logand number #x0f)
396 (logand (buffer-unsigned-byte-8 (frame-header frame) 5) #xf0)))))
397
398
399 (defgeneric frame-size (frame)
400 (:method ((frame 7-byte-header-frame))
401 (buffer-unsigned-byte-32 (frame-header frame) 3))
402
403 (:method ((frame 8-byte-header-frame))
404 (buffer-unsigned-byte-32 (frame-header frame) 4))
405
406 (:method ((frame 12-byte-header-frame))
407 (buffer-unsigned-byte-16 (frame-header frame) 2)))
408
409 (defgeneric setf-frame-size (size frame)
410 (:method (size (frame 7-byte-header-frame))
411 (setf (buffer-unsigned-byte-32 (frame-header frame) 3) size))
412
413 (:method (size (frame 8-byte-header-frame))
414 (setf (buffer-unsigned-byte-32 (frame-header frame) 4) size))
415
416 (:method (size (frame 12-byte-header-frame))
417 (setf (buffer-unsigned-byte-16 (frame-header frame) 2) size)))
418
419
420 (defgeneric frame-class-code (frame)
421 (:method ((frame 7-byte-header-frame))
422 (buffer-unsigned-byte-16 (frame-data frame) 0))
423
424 (:method ((frame 8-byte-header-frame))
425 (buffer-unsigned-byte-16 (frame-data frame) 0))
426
427 (:method ((frame 12-byte-header-frame))
428 (buffer-unsigned-byte-8 (frame-data frame) 0)))
429
430 (defgeneric setf-frame-class-code (code frame)
431 (:method (code (frame 7-byte-header-frame))
432 (setf (buffer-unsigned-byte-16 (frame-data frame) 0) code))
433
434 (:method (code (frame 8-byte-header-frame))
435 (setf (buffer-unsigned-byte-16 (frame-data frame) 0) code))
436
437 (:method (code (frame 12-byte-header-frame))
438 (setf (buffer-unsigned-byte-8 (frame-data frame) 0) code)))
439
440
441 (defgeneric frame-class-name (frame)
442 (:method ((frame amqp::frame))
443 (connection-class-code-class-name (frame-connection frame) (frame-class-code frame))))
444
445 (defgeneric (setf frame-class-name) (name frame)
446 (:method (name (frame amqp::frame))
447 (setf-frame-class-code (connection-class-name-class-code (frame-connection frame) name)
448 frame)))
449
450
451 (defgeneric frame-method-code (frame)
452 (:method ((frame 7-byte-header-frame))
453 (buffer-unsigned-byte-16 (frame-data frame) 2))
454
455 (:method ((frame 8-byte-header-frame))
456 (buffer-unsigned-byte-16 (frame-data frame) 2))
457
458 (:method ((frame 12-byte-header-frame))
459 (buffer-unsigned-byte-8 (frame-data frame) 1)))
460
461 (defgeneric setf-frame-method-code (code frame)
462 (:method (code (frame 7-byte-header-frame))
463 (setf (buffer-unsigned-byte-16 (frame-data frame) 2) code))
464
465 (:method (code (frame 8-byte-header-frame))
466 (setf (buffer-unsigned-byte-16 (frame-data frame) 2) code))
467
468 (:method (code (frame 12-byte-header-frame))
469 (setf (buffer-unsigned-byte-8 (frame-data frame) 1) code)))
470
471 (defgeneric frame-method-name (frame)
472 (:method ((frame amqp::frame))
473 (connection-method-code-method-name (frame-connection frame) (frame-class-code frame) (frame-method-code frame))))
474
475 (defgeneric (setf frame-method-name) (name frame)
476 (:method ((name symbol) (frame amqp::frame))
477 (setf-frame-method-code (connection-method-name-method-code (frame-connection frame)
478 (frame-class-code frame)
479 name)
480 frame)))
481
482
483 (defgeneric content-header-class-id (frame)
484 (:method ((buffer vector))
485 (buffer-unsigned-byte-16 buffer 0))
486 (:method ((frame amqp:frame))
487 (buffer-unsigned-byte-16 (frame-data frame) 0)))
488
489 (defgeneric setf-content-header-class-id (id frame)
490 (:method (id (buffer vector))
491 (setf (buffer-unsigned-byte-16 buffer 0) id))
492 (:method (id (frame amqp:frame))
493 (setf (buffer-unsigned-byte-16 (frame-data frame) 0) id)))
494
495 (defgeneric content-header-class-name (frame)
496 (:method ((frame amqp::frame))
497 (connection-class-code-class-name (frame-connection frame) (content-header-class-id frame))))
498
499
500 (defgeneric content-weight (frame)
501 (:method ((buffer vector))
502 (buffer-unsigned-byte-16 buffer 2))
503 (:method ((frame amqp:frame))
504 (buffer-unsigned-byte-16 (frame-data frame) 2)))
505
506
507 (defgeneric setf-content-weight (id frame)
508 (:method (id (buffer vector))
509 (setf (buffer-unsigned-byte-16 buffer 2) id))
510 (:method (id (frame amqp:frame))
511 (setf (buffer-unsigned-byte-16 (frame-data frame) 2) id)))
512
513
514 (defgeneric content-body-size (frame)
515 (:method ((buffer vector))
516 (buffer-unsigned-byte-64 buffer 4))
517 (:method ((frame amqp:frame))
518 (buffer-unsigned-byte-64 (frame-data frame) 4)))
519
520
521 (defgeneric setf-content-body-size (id frame)
522 (:method (id (buffer vector))
523 (setf (buffer-unsigned-byte-64 buffer 4) id))
524 (:method (id (frame amqp:frame))
525 (setf (buffer-unsigned-byte-64 (frame-data frame) 4) id)))
526
527
528
529
530 ;;;
531 ;;; frame initialization -
532 ;;; leave them as individually specialized until know if there is some reliable system...
533
534 (defmethod initialize-instance ((instance 8-byte-header-frame)
535 &rest initargs
536 &key
537 connection
538 ;; maximum payload is frame max net both the header
539 ;; _and_ the end byte (cf. http://dev.rabbitmq.com/wiki/Amqp08To091)
540 ;; in this case, 8 byte header + 1 end byte
541 (frame-size (connection-frame-size connection))
542 (header (make-array 8 :element-type '(unsigned-byte 8)))
543 (data (make-array (- frame-size 9) :element-type '(unsigned-byte 8))))
544 (declare (dynamic-extent initargs))
545 (apply #'call-next-method instance
546 :header header
547 :data data
548 initargs))
549
550 (defmethod initialize-instance ((instance 7-byte-header-frame)
551 &rest initargs
552 &key
553 connection
554 ;; maximum payload is frame max net both the header
555 ;; _and_ the end byte (cf. http://dev.rabbitmq.com/wiki/Amqp08To091)
556 ;; in this case, 7 byte header + 1 end byte
557 (frame-size (connection-frame-size connection))
558 (header (make-array 7 :element-type '(unsigned-byte 8)))
559 (data (make-array (- frame-size 8) :element-type '(unsigned-byte 8))))
560 (declare (dynamic-extent initargs))
561 (apply #'call-next-method instance
562 :header header
563 :data data
564 initargs))
565
566 (defmethod initialize-instance ((instance 12-byte-header-frame)
567 &rest initargs
568 &key
569 connection
570 (frame-size (connection-frame-size connection))
571 (header (make-array 12 :element-type '(unsigned-byte 8)))
572 (data (make-array (- frame-size 12) :element-type '(unsigned-byte 8))))
573 (declare (dynamic-extent initargs))
574 (apply #'call-next-method instance
575 :header header
576 :data data
577 initargs))
578
579 ;;;
580 ;;; frame input operators
581
582
583 (defgeneric read-frame (connection frame &key start end)
584 (:documentation "Read from the connection socket into the given frame.
585 This varies per protocol as the header layout varies.")
586
587 #+amqp.log-frames
588 (:method :around ((connection amqp:connection) (frame t) &key start end)
589 (multiple-value-prog1 (call-next-method)
590 (amqp:log :debug connection "read-frame: (~a,~a) ~s" start end frame)))
591
592 (:method :before ((connection amqp:connection) (frame t) &key start end)
593 (declare (ignore start end))
594 (unless (open-stream-p connection)
595 (error 'end-of-file :stream connection)))
596
597 (:method :after ((connection amqp:connection) (frame t) &key start end)
598 (declare (ignore start end))
599 (setf (connection-read-frame-timestamp connection) (get-universal-time))))
600
601 (defgeneric write-frame (connection frame &key start end)
602 (:documentation "Write from the given frame to the connection socket.
603 This varies per protocol as the header layout varies.")
604
605 #+amqp.log-frames
606 (:method :around ((connection amqp:connection) (frame t) &key start end)
607 (amqp:log :debug connection "write-frame: (~a,~a) ~s" start end frame)
608 (call-next-method))
609
610 (:method ((channel amqp:channel) (frame amqp:frame) &rest args)
611 (declare (dynamic-extent args))
612 (apply #'write-frame (channel-connection channel) frame args))
613
614 (:method :after ((connection amqp:connection) (frame t) &key start end)
615 (declare (ignore start end))
616 (setf (connection-write-frame-timestamp connection) (get-universal-time))))
617
618
619 (defmethod read-frame ((connection amqp:connection) (frame amqp.u:7-byte-header-input-frame) &rest args)
620 (declare (dynamic-extent args))
621 (apply #'read-7-byte-header-frame frame (stream-input-handle connection)
622 args))
623
624 (defmethod write-frame ((connection amqp:connection) (frame amqp.u:7-byte-header-output-frame) &key (start 0) (end nil))
625 (let ((stream (stream-output-handle connection)))
626 (setf end (or end (frame-size frame)))
627 (write-sequence (frame-header frame) stream :start start)
628 (write-sequence (frame-data frame) stream :end end)
629 (write-sequence #(#xCE) stream)
630 (force-output stream)
631 (+ end 7 1))) ; total count
632
633
634 (defmethod read-frame ((connection amqp:connection) (frame amqp.u:8-byte-header-input-frame) &rest args)
635 (declare (dynamic-extent args))
636 (apply #'read-8-byte-header-frame frame (stream-input-handle connection)
637 args))
638
639 (defmethod write-frame ((connection amqp:connection) (frame amqp.u:8-byte-header-output-frame) &key (start 0) (end nil))
640 (let ((stream (stream-output-handle connection)))
641 (setf end (or end (frame-size frame)))
642 (write-sequence (frame-header frame) stream :start start)
643 (write-sequence (frame-data frame) stream :end end)
644 (write-sequence #(#xCE) stream)
645 (force-output stream)
646 (+ end 1 8))) ; total count
647
648
649 (defmethod read-frame ((connection amqp:connection) (frame amqp.u:12-byte-header-input-frame) &rest args)
650 (declare (dynamic-extent args))
651 (apply #'read-12-byte-header-frame frame (stream-input-handle connection)
652 args))
653
654 (defmethod write-frame ((connection amqp:connection) (frame amqp.u:12-byte-header-output-frame) &key (start 0) (end nil))
655 (let ((stream (stream-output-handle connection)))
656 (setf end (or end (frame-size frame)))
657 (write-sequence (frame-header frame) stream :start start)
658 (write-sequence (frame-data frame) stream :end end)
659 (+ end 12)))
660
661
662 (document (read-7-byte-header-frame read-8-byte-header-frame read-12-byte-header-frame)
663 "The abstract frame structure (header . payload), is implemented variously in the respective versions.
664 The header size varies. The end marker is eliminated. The field sizes change. To allow for this, each
665 connection uses a specialized frame class and that class implements the general frame format.
666 This includes the buffer access as well as mediating translation between protocol-specific wire codes
667 for operations and classes and the generic names. Each protocol implements its logic in its
668 'abstract-classes' and 'classes' files.
669 nb. some stream classes which appear as handles may not specialize the respective `stream-read-*`
670 operators.")
671
672
673 (defun read-7-byte-header-frame (frame stream &key (start 0))
674 "Read a frame with a 7-byte header into a FRAME from a binary STREAM.
675 Permit a START offset to allow for the first already-read byte of an initial
676 start frame.
677 VALUES : the frame
678 the complete frame length = header + payload net end marker
679 CONDITIONS : signals end-of file if either the header or the payload is incomplete."
680 (unless (eql 7 (read-sequence (frame-header frame) stream :start start :end 7))
681 (error 'end-of-file :stream stream))
682 (let* ((payload-length (buffer-unsigned-byte-32 (frame-header frame) 3))
683 (frame-length (+ 7 payload-length))
684 (frame-end-marker nil)
685 (data (frame-data frame)))
686 ;; ignore the end marker for the length calculation
687 (assert (<= payload-length (length data)) ()
688 "Frame length exceeds buffer size: ~s, ~s"
689 payload-length data)
690 (unless (eql payload-length (read-sequence data stream :end payload-length))
691 (error 'end-of-file :stream stream))
692 (assert (eql (setf frame-end-marker (read-byte stream)) #xCE) ()
693 "Invalid frame end: ~s" frame-end-marker)
694
695 (values frame frame-length)))
696
697
698 (defun read-8-byte-header-frame (frame stream &key (start 0))
699 "Read a frame with a 8-byte header into a FRAME from a binary STREAM.
700 Permit a START offset to allow for the first already-read byte of an iitial
701 start frame.
702 VALUES : the frame
703 the complete frame length = header + payload net end marker
704 CONDITIONS : signals end-of file if either the header or the payload is incomplete."
705 (unless (eql 8 (read-sequence (frame-header frame) stream :start start :end 8))
706 (error 'end-of-file :stream stream))
707 (let* ((payload-length (buffer-unsigned-byte-32 (frame-header frame) 4))
708 (frame-length (+ 8 payload-length))
709 (frame-end-marker nil)
710 (data (frame-data frame)))
711 ;; ignore the end marker for the length calculation
712 (assert (<= payload-length (length (frame-data frame))) ()
713 "Frame length exceeds buffer size: ~s, ~s"
714 payload-length data)
715 (unless (eql payload-length (read-sequence data stream :end payload-length))
716 (error 'end-of-file :stream stream))
717 (assert (eql (setf frame-end-marker (read-byte stream)) #xCE) ()
718 "Invalid frame end: ~s" frame-end-marker)
719 (values frame frame-length)))
720
721 (defun read-12-byte-header-frame (frame stream &key (start 0))
722 "Read a frame with a 12-byte header into a BUFFER from a binary STREAM.
723 Permit a START offset to allow for the first already-read byte of an iitial
724 start frame.
725 VALUES : the frame
726 the complete frame length = header + payload net end marker
727 CONDITIONS : signals end-of file if either the header or the payload is incomplete."
728 (unless (eql 12 (read-sequence (frame-header frame) stream :start start :end 12))
729 (error 'end-of-file :stream stream))
730 (let* ((frame-length (buffer-unsigned-byte-16 (frame-header frame) 2))
731 (payload-length (- frame-length 12))
732 (data (frame-data frame)))
733 (assert (<= payload-length (length data)) ()
734 "Frame length exceeds buffer size: ~s, ~s"
735 payload-length (length data))
736 (unless (eql payload-length (read-sequence data stream :end payload-length))
737 (error 'end-of-file :stream stream))
738 (values frame frame-length)))

  ViewVC Help
Powered by ViewVC 1.1.5