Return error when stream source fails (Fixes #11)

chunker.go

@@ -126,7 +126,7 @@ func getBoundary(resp *http.Response) (string, error) {
 		return "", err
 	}
 	if !strings.HasPrefix(mediaType, "multipart/") {
-		return "", fmt.Errorf("expected multipart media type: %s", contentType)
+		return "", fmt.Errorf("unexpected media type: %s", contentType)
 	}
 
 	boundary := params["boundary"]

pubsub.go

@@ -193,9 +193,10 @@ func clientAddress(r *http.Request) string {
 func (pubSub *PubSub) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	if r.Method != http.MethodGet && r.Method != http.MethodHead {
 		w.Header().Set("Allow", fmt.Sprintf("%s, %s", http.MethodGet, http.MethodHead))
-		http.Error(w, "405 Method Not Allowed", http.StatusMethodNotAllowed)
+		http.Error(w, fmt.Sprintf("HTTP method %s not supported", r.Method), http.StatusMethodNotAllowed)
 		return
 	}
+
 	// prepare response for flushing
 	flusher, ok := w.(http.Flusher)
 	if !ok {
@@ -255,6 +256,12 @@ LOOP:
 		flusher.Flush()
 	}
 
+	if !headersSent && !chunkOk {
+		fmt.Printf("server[%s]: stream failed\n", pubSub.id)
+		http.Error(w, "Stream failed", http.StatusServiceUnavailable)
+		return
+	}
+
 	err := mw.Close()
 	if err != nil {
 		fmt.Printf("server[%s]: mime close failed: %s\n", pubSub.id, err)