Queue flushing

hub/hub.go

@@ -17,6 +17,10 @@ type (
 		Queue   string
 		Message []byte
 	}
+	MessageDump struct {
+		Queue   string `json:"queue"`
+		Message string `json:"message"`
+	}
 )
 
 func New(st *storage.Storage) *Hub {
@@ -61,6 +65,20 @@ func (h *Hub) Sub(s *Subscription) {
 	h.subscribers = append(h.subscribers, s)
 }
 
+func (h *Hub) Flush(queues []string) (messages []MessageDump) {
+	for _, queue := range queues {
+		for _, msg := range h.storage.Flush(queue) {
+			messages = append(messages, MessageDump{queue, string(msg)})
+		}
+	}
+
+	if messages == nil {
+		messages = []MessageDump{}
+	}
+
+	return
+}
+
 func (h *Hub) Info() map[string]map[string]uint {
 	info := make(map[string]map[string]uint)
 

server/server.go

@@ -32,6 +32,7 @@ func New(port int, h *hub.Hub) *Server {
 	http.HandleFunc("/debug", s.debugHandler)
 	http.HandleFunc("/publish", s.pubHandler)
 	http.HandleFunc("/subscribe", s.subHandler)
+	http.HandleFunc("/flush", s.flushHandler)
 
 	return &s
 }
@@ -97,3 +98,11 @@ func (s *Server) subHandler(w http.ResponseWriter, r *http.Request) {
 		w.Write(res.Message)
 	}
 }
+
+func (s *Server) flushHandler(w http.ResponseWriter, r *http.Request) {
+	queues := strings.Split(r.FormValue("queues"), ",")
+	messages := s.hub.Flush(queues)
+
+	jsn, _ := json.Marshal(messages)
+	w.Write(jsn)
+}

storage/storage.go

@@ -78,6 +78,20 @@ func (s *Storage) Put(queue string, message []byte) (err error) {
 	return
 }
 
+func (s *Storage) Flush(queue string) (messages [][]byte) {
+	done := make(chan struct{})
+
+	for {
+		if msg, ok := s.Get(queue, done); ok {
+			messages = append(messages, msg)
+		} else {
+			return
+		}
+	}
+
+	return
+}
+
 func (s *Storage) QueueSizes() map[string]uint {
 	info := make(map[string]uint)