<commit_before>(in-package :cl-async)

(defclass async-stream (trivial-gray-stream-mixin)
  ((socket :accessor stream-socket :initarg :socket :initform nil)
   (buffer :accessor stream-buffer :initform (make-buffer)))
  (:documentation "The underlying class for async streams. Wraps a tcp socket class."))
(defclass async-output-stream (async-stream fundamental-binary-output-stream) ()
  (:documentation "Async output stream."))
(defclass async-input-stream (async-stream fundamental-binary-input-stream) ()
  (:documentation "Async input stream."))
(defclass async-io-stream (async-input-stream async-output-stream) ()
  (:documentation "Async stream for both input and output."))

;; -----------------------------------------------------------------------------
;; base stream
;; -----------------------------------------------------------------------------
(defmethod stream-append-bytes ((stream async-stream) bytes)
  "Append some data to a stream's underlying buffer."
  (write-to-buffer bytes (stream-buffer stream)))

(defmethod stream-output-type ((stream async-stream))
  "This is always a binary stream."
  'cl-async-util:octet)

(defmethod stream-element-type ((stream async-stream))
  "This is always a binary stream."
  'cl-async-util:octet)

(defmethod open-stream-p ((stream async-stream))
  "Test the underlying socket to see if this stream is open."
  (let ((socket (stream-socket stream)))
    (not (as:socket-closed-p socket))))

(defmethod close ((stream async-stream) &key abort)
  "Close the stream. If aborting, attempt to clear out remaining data in the
   buffers before closing (is this really needed?)"
  (when abort
    (when (output-stream-p stream)
      (clear-output stream))
    (when (input-stream-p stream)
      (clear-input stream)))
  (as:close-socket (stream-socket stream)))

;; -----------------------------------------------------------------------------
;; output stream
;; -----------------------------------------------------------------------------
(defmethod stream-clear-output ((stream async-output-stream))
  "Attempt to clear the output buffer of an output stream."
  ;; we don't have the concept of an output buffer, only an underlying UV stream
  ;; that's either written to or isn't
  nil)

(defmethod stream-force-output ((stream async-output-stream))
  "Force an output stream to send its data to the underlying fd."
  ;; we don't have the concept of an output buffer, only an underlying UV stream
  ;; that's either written to or isn't
  nil)

(defmethod stream-finish-output ((stream async-output-stream))
  "Really, since we're async, same as force-output."
  (stream-force-output stream))

(defmethod stream-write-sequence ((stream async-output-stream) sequence start end &key)
  "Write a sequence of bytes to the underlying socket."
  (when (open-stream-p stream)
    (let ((seq (subseq sequence start end)))
      (write-socket-data (stream-socket stream) seq))))

(defmethod stream-write-byte ((stream async-output-stream) byte)
  "Write one byte to the underlying socket."
  (stream-write-sequence stream (vector byte) 0 1))
  
(defmethod send-buffered-data ((stream async-output-stream))
  "Take data we've buffered between initial sending and actual socket connection
   and send it out."
  (let ((data (buffer-output (stream-buffer stream))))
    (setf (stream-buffer stream) (make-buffer))
    (write-socket-data (stream-socket stream) data))
  nil)

;; -----------------------------------------------------------------------------
;; input stream
;; -----------------------------------------------------------------------------
(defmethod stream-clear-input ((stream async-input-stream))
  "Attempt to clear the input buffer of an input stream."
  (when (open-stream-p stream)
    (setf (stream-buffer stream) (make-buffer))))

(defmethod stream-read-byte ((stream async-input-stream))
  "Read one byte from the underlying socket."
  (let* ((buff (make-array 1 :element-type '(unsigned-byte 8)))
         (num (stream-read-sequence stream buff 0 1)))
    (if (= num 1)
        (aref buff 0)
        :eof)))

(defmethod stream-read-sequence ((stream async-input-stream) sequence start end &key)
  "Attempt to read a sequence of bytes from the underlying socket."
  (let* ((buffer (buffer-output (stream-buffer stream)))
         (numbytes (min (length buffer) (- end start)))
         (bytes (subseq buffer start (min (length buffer) numbytes))))
    (setf (stream-buffer stream) (make-buffer (subseq buffer numbytes)))
    (replace sequence bytes)
    (length bytes)))

<commit_msg>Fix write-sequence to tcp stream on sbcl.<commit_after>(in-package :cl-async)

(defclass async-stream (trivial-gray-stream-mixin)
  ((socket :accessor stream-socket :initarg :socket :initform nil)
   (buffer :accessor stream-buffer :initform (make-buffer)))
  (:documentation "The underlying class for async streams. Wraps a tcp socket class."))
(defclass async-output-stream (async-stream fundamental-binary-output-stream) ()
  (:documentation "Async output stream."))
(defclass async-input-stream (async-stream fundamental-binary-input-stream) ()
  (:documentation "Async input stream."))
(defclass async-io-stream (async-input-stream async-output-stream) ()
  (:documentation "Async stream for both input and output."))

;; -----------------------------------------------------------------------------
;; base stream
;; -----------------------------------------------------------------------------
(defmethod stream-append-bytes ((stream async-stream) bytes)
  "Append some data to a stream's underlying buffer."
  (write-to-buffer bytes (stream-buffer stream)))

(defmethod stream-output-type ((stream async-stream))
  "This is always a binary stream."
  'cl-async-util:octet)

(defmethod stream-element-type ((stream async-stream))
  "This is always a binary stream."
  'cl-async-util:octet)

(defmethod open-stream-p ((stream async-stream))
  "Test the underlying socket to see if this stream is open."
  (let ((socket (stream-socket stream)))
    (not (as:socket-closed-p socket))))

(defmethod close ((stream async-stream) &key abort)
  "Close the stream. If aborting, attempt to clear out remaining data in the
   buffers before closing (is this really needed?)"
  (when abort
    (when (output-stream-p stream)
      (clear-output stream))
    (when (input-stream-p stream)
      (clear-input stream)))
  (as:close-socket (stream-socket stream)))

;; -----------------------------------------------------------------------------
;; output stream
;; -----------------------------------------------------------------------------
(defmethod stream-clear-output ((stream async-output-stream))
  "Attempt to clear the output buffer of an output stream."
  ;; we don't have the concept of an output buffer, only an underlying UV stream
  ;; that's either written to or isn't
  nil)

(defmethod stream-force-output ((stream async-output-stream))
  "Force an output stream to send its data to the underlying fd."
  ;; we don't have the concept of an output buffer, only an underlying UV stream
  ;; that's either written to or isn't
  nil)

(defmethod stream-finish-output ((stream async-output-stream))
  "Really, since we're async, same as force-output."
  (stream-force-output stream))

(defmethod stream-write-sequence ((stream async-output-stream) sequence start end &key)
  "Write a sequence of bytes to the underlying socket."
  (when (open-stream-p stream)
    (let ((seq (subseq sequence start end)))
      (write-socket-data (stream-socket stream) seq)
      seq)))

(defmethod stream-write-byte ((stream async-output-stream) byte)
  "Write one byte to the underlying socket."
  (stream-write-sequence stream (vector byte) 0 1))

(defmethod send-buffered-data ((stream async-output-stream))
  "Take data we've buffered between initial sending and actual socket connection
   and send it out."
  (let ((data (buffer-output (stream-buffer stream))))
    (setf (stream-buffer stream) (make-buffer))
    (write-socket-data (stream-socket stream) data))
  nil)

;; -----------------------------------------------------------------------------
;; input stream
;; -----------------------------------------------------------------------------
(defmethod stream-clear-input ((stream async-input-stream))
  "Attempt to clear the input buffer of an input stream."
  (when (open-stream-p stream)
    (setf (stream-buffer stream) (make-buffer))))

(defmethod stream-read-byte ((stream async-input-stream))
  "Read one byte from the underlying socket."
  (let* ((buff (make-array 1 :element-type '(unsigned-byte 8)))
         (num (stream-read-sequence stream buff 0 1)))
    (if (= num 1)
        (aref buff 0)
        :eof)))

(defmethod stream-read-sequence ((stream async-input-stream) sequence start end &key)
  "Attempt to read a sequence of bytes from the underlying socket."
  (let* ((buffer (buffer-output (stream-buffer stream)))
         (numbytes (min (length buffer) (- end start)))
         (bytes (subseq buffer start (min (length buffer) numbytes))))
    (setf (stream-buffer stream) (make-buffer (subseq buffer numbytes)))
    (replace sequence bytes)
    (length bytes)))

{'fraction_non_alphanumeric': 0.21507493709659775, 'fraction_numerical': 0.0017503555409692593, 'mean_word_length': 2.5724892536146933, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '605041', 'n_tokens_mistral': 2444, 'n_tokens_neox': 2371, 'n_words': 873}