xyttrでstreaming-apiを使用 (とりあえずfilter.jsonのtrackのみ)

xyttr-streaming-api.l

;;; -*- mode: lisp; package: xyttr -*-

(eval-when (:compile-toplevel :load-toplevel :execute)
  (require "cmu_loop")
  (require "json")
  (require "xyttr"))
  
(in-package :xyttr)

(defconstant *stream-api-host* "stream.twitter.com")
(defconstant *stream-api-method* "/1/statuses/filter.json")

(defvar *streaming-connection* nil)
(defvar *streaming-buffer* nil)
(defvar *streaming-polling-interval* 0.2)

(defun url-encode (e)
  (si:www-url-encode
   (convert-encoding-from-internal *encoding-utf8n* (format nil "~A" e))))

;; ref. https://dev.twitter.com/docs/streaming-api/methods
(defun streaming-connect (params &key user onsuccess onfailure)
  (let* ((cn (connect *stream-api-host* 80 :encoding :text))
	 (path (format nil "~A?~{~A=~A~^&~}" *stream-api-method* (mapcar #'url-encode params)))
	 (cred (list :consumer-key *consumer-key* :consumer-secret *consumer-secret*
		     :token *token* :token-secret *token-secret*))
	 (auth-url (format nil "http://~A~A" *stream-api-host* *stream-api-method*))
	 (auth (oauth:auth-header cred 'get auth-url params)))
    (setq *streaming-connection* cn)
    (message "Connecting stream.twitter.com ...")
    (format cn "GET ~A HTTP/1.1~%~
                Host: stream.twitter.com~%~
                Connection: Keep-Alive~%~
                User-Agent: xyzzy/~A~%" path (software-version))
    (format cn "Authorization: ~A~%~%" auth)
    (labels
	((listener ()
	   (when (listen cn)
	     (stop-timer #'listener)
	     (let ((head (read-line cn)))
	       (when (not (string-match "200" head))
		 (let ((res (cons head (loop while (listen cn) collect (read-line cn nil)))))
		   (close cn)
		   (if onfailure
		       (funcall onfailure res)
		     (message-box (format nil "~{~A~%~}" res) "接続失敗" (list :ok :exclamation)))))
	       (message "Connected. OK")
	       (while (string/= (read-line cn nil) ""))
	       (setq *streaming-connection* cn)
	       (funcall onsuccess)))))
      (start-timer 0.5 #'listener))
    t))

(defun streaming-close ()
  (when (and *streaming-connection*
	     (open-stream-p *streaming-connection*))
    (close *streaming-connection*))
  (setq *streaming-connection* nil))



(defun streaming-receive-statuses ()
  (let ((statuses nil))
    (while (listen #0=*streaming-connection*)
      (let ((l (read-line #0# nil)))
	(unless (string-match "^[ 0-9A-F]*$" l)
	  (push (json:json-decode l) statuses))))
    (nreverse statuses)))


(defun streaming-start (params buf &key (user *default-user*))
  (set-buffer buf)
  (streaming-connect
    params
    :user user
    :onfailure
    (lambda (res)
      (message-box (format nil "~{~A~%~}" res) "接続失敗" (list :ok :exclamation))
      (close-timeline-buffer buf)
      (refresh-screen))
    :onsuccess
    (lambda ()
      (set-buffer buf)
      (setq *streaming-buffer* buf)
      (labels
	  ((listener ()
	     (if (deleted-buffer-p buf)
		 (stop-timer #'listener)
	       (whenlet res (streaming-receive-statuses)
		 (timeline-draw-statuses buf res)
		 (refresh-screen))))
	   (closer (b)
	     (when (eq b buf)
	       (stop-timer #'listener)
	       (delete-hook '*before-delete-buffer-hook* #'closer)
	       (streaming-close)
	       (setq *streaming-buffer* nil))
	     t))
	(start-timer 1 #'listener)
	(add-hook '*before-delete-buffer-hook* #'closer))))
   t)


;;; command
(defun user::xyttr-stream-track (track)
  (interactive "sTrack: ")
  (when (and *streaming-connection*
	     (open-stream-p *streaming-connection*))
    (if (eq (message-box (format nil "~Aでのストリーミングを終了しますか?"
				 (buffer-name *streaming-buffer*))
			 "確認" (list :question :yes-no)) :yes)
	(streaming-close)
      (quit)))
  (setq *streaming-connection* nil
	*streaming-buffer* nil)
  (let ((buf (get-buffer-create (format nil "*tws*: ~A" track))))
    (when *use-frame*
      (select-pseudo-frame
       (or (find-pseudo-frame *frame-name*)
	   (new-pseudo-frame *frame-name*))))
    (set-buffer buf)
    (xyttr-timeline-mode)
    (setf (modeconf-mode buffer-modeconf) :xyttr-stream-track)
    (streaming-start `(:track ,track) buf :user *default-user*)))