feat: SSE add request failure event callback function #1041 (#1076)

Author: jeevatkm

sse.go

@@ -51,6 +51,10 @@ type (
 	// when an error occurs with [EventSource] processing
 	EventErrorFunc func(error)
 
+	// EventRequestFailureFunc is a callback function type used to receive event
+	// details from the Server-Sent Events(SSE) request failure
+	EventRequestFailureFunc func(err error, res *http.Response)
+
 	// Event struct represents the event details from the Server-Sent Events(SSE) stream
 	Event struct {
 		ID   string
@@ -76,6 +80,7 @@ type (
 		maxBufSize       int
 		onOpen           EventOpenFunc
 		onError          EventErrorFunc
+		onRequestFailure EventRequestFailureFunc
 		onEvent          map[string]*callback
 		log              Logger
 		closed           bool
@@ -343,12 +348,33 @@ func (es *EventSource) OnError(ef EventErrorFunc) *EventSource {
 	defer es.lock.Unlock()
 	if es.onError != nil {
 		es.log.Warnf("Overwriting an existing OnError callback from=%s to=%s",
-			functionName(es.OnError), functionName(ef))
+			functionName(es.onError), functionName(ef))
 	}
 	es.onError = ef
 	return es
 }
 
+// OnRequestFailure registered callback gets triggered when the HTTP request
+// failure while establishing a SSE connection.
+//
+//	es.OnRequestFailure(func(err error, res *http.Response) {
+//		fmt.Println("Error and response:", err, res)
+//	})
+//
+// Note:
+//   - Do not forget to close the HTTP response body.
+//   - HTTP response may be nil.
+func (es *EventSource) OnRequestFailure(ef EventRequestFailureFunc) *EventSource {
+	es.lock.Lock()
+	defer es.lock.Unlock()
+	if es.onRequestFailure != nil {
+		es.log.Warnf("Overwriting an existing OnRequestFailure callback from=%s to=%s",
+			functionName(es.onRequestFailure), functionName(ef))
+	}
+	es.onRequestFailure = ef
+	return es
+}
+
 // OnMessage method registers a callback to emit every SSE event message
 // from the server. The second result argument is optional; it can be used
 // to register the data type for JSON data.
@@ -495,6 +521,14 @@ func (es *EventSource) triggerOnError(err error) {
 	}
 }
 
+func (es *EventSource) triggerOnRequestFailure(err error, res *http.Response) {
+	es.lock.RLock()
+	defer es.lock.RUnlock()
+	if es.onRequestFailure != nil {
+		es.onRequestFailure(err, res)
+	}
+}
+
 func (es *EventSource) createRequest() (*http.Request, error) {
 	req, err := http.NewRequest(es.method, es.url, es.body)
 	if err != nil {
@@ -557,6 +591,9 @@ func (es *EventSource) connect() (*http.Response, error) {
 			} else {
 				err = doErr
 			}
+			if err != nil {
+				es.triggerOnRequestFailure(err, resp)
+			}
 			break
 		}
 

sse_test.go

@@ -342,6 +342,31 @@ func TestGH1044TrimHeader(t *testing.T) {
 	})
 }
 
+func TestGH1041RequestFailureWithResponseBody(t *testing.T) {
+	es := createEventSource(t, "", func(any) {}, nil)
+	ts := createTestServer(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set(hdrContentTypeKey, jsonContentType)
+		w.WriteHeader(http.StatusBadRequest)
+		_, _ = w.Write([]byte(`{ "id": "bad_request", "message": "Unable to establish connection" }`))
+	})
+	defer ts.Close()
+
+	rfFunc := func(err error, res *http.Response) {
+		defer res.Body.Close()
+		resBytes, _ := io.ReadAll(res.Body)
+
+		assertNotNil(t, err)
+		assertEqual(t, "resty:sse: 400 Bad Request", err.Error())
+		assertEqual(t, `{ "id": "bad_request", "message": "Unable to establish connection" }`, string(resBytes))
+	}
+
+	es.SetURL(ts.URL).OnRequestFailure(rfFunc)
+	es.OnRequestFailure(rfFunc)
+	err := es.Get()
+	assertNotNil(t, err)
+	assertEqual(t, "resty:sse: 400 Bad Request", err.Error())
+}
+
 func TestEventSourceHTTPError(t *testing.T) {
 	es := createEventSource(t, "", func(any) {}, nil)
 	ts := createTestServer(func(w http.ResponseWriter, r *http.Request) {