gistfile1.txt

func natsToHTTP(msg jetstream.Msg) *http.Request {
	hdrs := msg.Headers()
	request := http.Request{
		URL: &url.URL{
			Scheme: "queue",
			Path:   hdrs.Get(jetURLPath),
		},
		Method: http.MethodPost,
		Header: http.Header(hdrs),
		Body:   io.NopCloser(bytes.NewReader(msg.Data())),
	}

	return &request
}

jetstreamgo

func messageProcessor(ctx context.Context, mux *http.ServeMux, worker chan jetstream.Msg) {
	log := logkit.FromContext(ctx)
	for {
		select {
		case <-ctx.Done():
			return
		case msg := <-worker:
			req := natsToHTTP(msg)

			w := httptest.NewRecorder()
			mux.ServeHTTP(w, req.WithContext(ctx))

			res := w.Result()
			defer res.Body.Close()
			if res.StatusCode != http.StatusOK {
				buf, _ := io.ReadAll(io.LimitReader(res.Body, 1024))
				log.With("statusCode", res.StatusCode).With("statusMsg", string(buf)).Info("nats consumer error")
			}

			_ = msg.Ack()
		}
	}
}