fix: adapter to new serverless connector

internal/core/plugin_manager/serverless_connector/connector.go

@@ -72,7 +72,7 @@ func FetchFunction(manifest plugin_entities.PluginDeclaration, checksum string)
 	}
 
 	if response.Error != "" {
-		return nil, fmt.Errorf("unexpected response from serverless connector: %s", response.Error)
+		return nil, fmt.Errorf("unexpected response from plugin controller: %s", response.Error)
 	}
 
 	if len(response.Items) == 0 {
@@ -132,7 +132,6 @@ func SetupFunction(
 				},
 			},
 		),
-		http_requests.HttpRaiseErrorWhenStreamDataNotMatch(true),
 	)
 	if err != nil {
 		return nil, err
@@ -145,7 +144,7 @@ func SetupFunction(
 		"func":   "SetupFunction",
 	}, func() {
 		defer response.Close()
-		serverless_connector_response.Async(func(chunk LaunchFunctionResponseChunk) {
+		if err := serverless_connector_response.Async(func(chunk LaunchFunctionResponseChunk) {
 			if chunk.State == LAUNCH_STATE_FAILED {
 				response.Write(LaunchFunctionResponse{
 					Event:   Error,
@@ -191,7 +190,12 @@ func SetupFunction(
 					Message: "Plugin launched",
 				})
 			}
-		})
+		}); err != nil {
+			response.Write(LaunchFunctionResponse{
+				Event:   Error,
+				Message: err.Error(),
+			})
+		}
 	})
 
 	return response, nil

internal/core/plugin_manager/serverless_connector/types.go

@@ -52,5 +52,5 @@ type LaunchFunctionFinalStageMessage struct {
 }
 
 func getFunctionFilename(manifest plugin_entities.PluginDeclaration, checksum string) string {
-	return fmt.Sprintf("%s-%s_%s@%s.difypkg", manifest.Author, manifest.Name, manifest.Version, checksum)
+	return fmt.Sprintf("%s@%s@%s@%s.difypkg", manifest.Author, manifest.Name, manifest.Version, checksum)
 }

internal/utils/http_requests/http_warpper.go

@@ -9,6 +9,7 @@ import (
 	"net/http"
 	"time"
 
+	"github.com/langgenius/dify-plugin-daemon/internal/utils/log"
 	"github.com/langgenius/dify-plugin-daemon/internal/utils/parser"
 	"github.com/langgenius/dify-plugin-daemon/internal/utils/routine"
 	"github.com/langgenius/dify-plugin-daemon/internal/utils/stream"
@@ -123,17 +124,27 @@ func RequestAndParseStream[T any](client *http.Client, url string, method string
 				continue
 			}
 
-			if bytes.HasPrefix(data, []byte("data: ")) {
+			if bytes.HasPrefix(data, []byte("data:")) {
 				// split
-				data = data[6:]
+				data = data[5:]
 			}
 
+			if bytes.HasPrefix(data, []byte("event:")) {
+				// TODO: handle event
+				continue
+			}
+
+			// trim space
+			data = bytes.TrimSpace(data)
+
 			// unmarshal
 			t, err := parser.UnmarshalJsonBytes[T](data)
 			if err != nil {
 				if raiseErrorWhenStreamDataNotMatch {
 					ch.WriteError(err)
 					break
+				} else {
+					log.Warn("stream data not match for %s, got %s", url, string(data))
 				}
 				continue
 			}