Add active publish logic to streams

internal/streams/handlers.go

@@ -73,3 +73,25 @@ func Location(url string) (string, error) {
 
 	return "", nil
 }
+
+// TODO: rework
+
+type ConsumerHandler func(url string) (core.Consumer, func(), error)
+
+var consumerHandlers = map[string]ConsumerHandler{}
+
+func HandleConsumerFunc(scheme string, handler ConsumerHandler) {
+	consumerHandlers[scheme] = handler
+}
+
+func GetConsumer(url string) (core.Consumer, func(), error) {
+	if i := strings.IndexByte(url, ':'); i > 0 {
+		scheme := url[:i]
+
+		if handler, ok := consumerHandlers[scheme]; ok {
+			return handler(url)
+		}
+	}
+
+	return nil, nil, errors.New("streams: unsupported scheme: " + url)
+}

internal/streams/publish.go

@@ -0,0 +1,19 @@
+package streams
+
+func (s *Stream) Publish(url string) error {
+	cons, run, err := GetConsumer(url)
+	if err != nil {
+		return err
+	}
+
+	if err = s.AddConsumer(cons); err != nil {
+		return err
+	}
+
+	go func() {
+		run()
+		s.RemoveConsumer(cons)
+	}()
+
+	return nil
+}

internal/streams/streams.go

@@ -172,6 +172,10 @@ func streamsHandler(w http.ResponseWriter, r *http.Request) {
 				} else {
 					api.ResponseJSON(w, stream)
 				}
+			} else if stream = Get(src); stream != nil {
+				if err := stream.Publish(dst); err != nil {
+					http.Error(w, err.Error(), http.StatusInternalServerError)
+				}
 			} else {
 				http.Error(w, "", http.StatusNotFound)
 			}