add buffer mechanism & bugfix

go.mod

@@ -3,7 +3,7 @@ module github.com/OpenFunction/dapr-proxy
 go 1.18
 
 require (
-	github.com/OpenFunction/functions-framework-go v0.0.0-20220925145105-86f7bcc9cf8c
+	github.com/OpenFunction/functions-framework-go v0.5.0
 	github.com/dapr/components-contrib v1.8.1-rc.1
 	github.com/dapr/dapr v1.8.3
 	github.com/dapr/go-sdk v1.5.0

go.sum

@@ -57,8 +57,8 @@ github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym
 github.com/NYTimes/gziphandler v0.0.0-20170623195520-56545f4a5d46/go.mod h1:3wb06e3pkSAbeQ52E9H9iFoQsEEwGN64994WTCIhntQ=
 github.com/NYTimes/gziphandler v1.1.1/go.mod h1:n/CVRwUEOgIxrgPvAQhUUr9oeUtvrhMomdKFjzJNB0c=
 github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU=
-github.com/OpenFunction/functions-framework-go v0.0.0-20220925145105-86f7bcc9cf8c h1:TFstl4SFFVZpd2+2yhVGjW15em/lV9NJ7NGoM6kaUHQ=
-github.com/OpenFunction/functions-framework-go v0.0.0-20220925145105-86f7bcc9cf8c/go.mod h1:ussM725MZuGmAH4PPmwGdHoDxIT4oSx7VK5Wiibg/No=
+github.com/OpenFunction/functions-framework-go v0.5.0 h1:s2L4PyazI8EoPzrfW0Es1esCSZS2SFXD6KNXyDKaJKc=
+github.com/OpenFunction/functions-framework-go v0.5.0/go.mod h1:+uYjTEYmn2uqIyViZtg9OF+bUNdjbkWNd7jrQWc7iEc=
 github.com/PuerkitoBio/purell v1.1.1 h1:WEQqlqaGbrPkxLJWfBwQmfEAE1Z7ONdDLqrN38tNFfI=
 github.com/PuerkitoBio/purell v1.1.1/go.mod h1:c11w/QuzBsJSee3cPx9rAFu61PvFxuPbtSwDGJws/X0=
 github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578 h1:d+Bc7a5rLufV/sSk/8dngufqelfh6jnri85riMAaF/M=

main.go

@@ -6,15 +6,17 @@ import (
 	"strconv"
 	"time"
 
-	proxyruntime "github.com/OpenFunction/dapr-proxy/pkg/runtime"
-	"github.com/OpenFunction/dapr-proxy/pkg/utils"
 	ofctx "github.com/OpenFunction/functions-framework-go/context"
 	"github.com/OpenFunction/functions-framework-go/framework"
+	"github.com/cenkalti/backoff/v4"
 	diag "github.com/dapr/dapr/pkg/diagnostics"
 	"github.com/dapr/dapr/pkg/modes"
 	"github.com/dapr/dapr/pkg/runtime"
 	"github.com/pkg/errors"
 	"k8s.io/klog/v2"
+
+	proxyruntime "github.com/OpenFunction/dapr-proxy/pkg/runtime"
+	"github.com/OpenFunction/dapr-proxy/pkg/utils"
 )
 
 const (
@@ -48,17 +50,20 @@ func main() {
 	port, _ := strconv.Atoi(funcContext.GetPort())
 	protocol := utils.GetEnvVar(protocolEnvVar, defaultAppProtocol)
 	config := &proxyruntime.Config{
-		Protocol: runtime.Protocol(protocol),
-		Host:     host,
-		Port:     port,
-		Mode:     modes.KubernetesMode,
+		Protocol:      runtime.Protocol(protocol),
+		Host:          host,
+		Port:          port,
+		Mode:          modes.KubernetesMode,
+		MaxBufferSize: 1000000,
 	}
 
 	FuncRuntime = proxyruntime.NewFuncRuntime(config, funcContext)
 	if err := FuncRuntime.CreateFuncChannel(); err != nil {
 		klog.Exit(err)
 	}
 
+	go FuncRuntime.ProcessEvents()
+
 	if err := fwk.Register(ctx, EventHandler); err != nil {
 		klog.Exit(err)
 	}
@@ -75,35 +80,37 @@ func EventHandler(ctx ofctx.Context, in []byte) (ofctx.Out, error) {
 		klog.V(4).Infof("Input: %s - Event Forwarding Elapsed: %vms", ctx.GetInputName(), elapsed)
 	}()
 
-	// Forwarding BindingEvent
+	c := ctx.GetNativeContext()
+
+	// Handle BindingEvent
 	bindingEvent := ctx.GetBindingEvent()
 	if bindingEvent != nil {
-		data, err := FuncRuntime.OnBindingEvent(ctx, bindingEvent)
-		if err != nil {
-			klog.Error(err)
-			return ctx.ReturnOnInternalError(), err
-		} else {
-			out := new(ofctx.FunctionOut)
-			out.WithData(data)
-			out.WithCode(ofctx.Success)
-			return out, nil
-		}
+		FuncRuntime.EnqueueBindingEvent(&c, bindingEvent)
 	}
 
-	// Forwarding TopicEvent
+	// Handle TopicEvent
 	topicEvent := ctx.GetTopicEvent()
 	if topicEvent != nil {
-		err := FuncRuntime.OnTopicEvent(ctx, topicEvent)
-		if err != nil {
-			klog.Error(err)
-			return ctx.ReturnOnInternalError(), err
-		} else {
-			out := new(ofctx.FunctionOut)
-			out.WithCode(ofctx.Success)
-			return out, nil
-		}
+		FuncRuntime.EnqueueTopicEvent(&c, topicEvent)
 	}
 
-	err := errors.New("Only Binding and Pubsub events are supported")
-	return ctx.ReturnOnInternalError(), err
+	var resp *proxyruntime.EventResponse
+	err := backoff.Retry(func() error {
+		resp = FuncRuntime.GetEventResponse(&c)
+		if resp == nil {
+			return errors.New("Failed to get event response")
+		}
+		return nil
+	}, utils.NewExponentialBackOff())
+
+	if err != nil {
+		e := errors.New("Processing event timeout")
+		klog.Error(e)
+		return ctx.ReturnOnInternalError(), e
+	} else {
+		out := new(ofctx.FunctionOut)
+		out.WithData(resp.Data)
+		out.WithCode(ofctx.Success)
+		return out, nil
+	}
 }

pkg/grpc/grpc.go

@@ -10,7 +10,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/OpenFunction/dapr-proxy/pkg/lb"
 	"github.com/dapr/dapr/pkg/channel"
 	diag "github.com/dapr/dapr/pkg/diagnostics"
 	"github.com/dapr/dapr/pkg/modes"
@@ -20,6 +19,8 @@ import (
 	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/credentials"
 	"k8s.io/klog/v2"
+
+	"github.com/OpenFunction/dapr-proxy/pkg/lb"
 )
 
 type endpoint string
@@ -89,7 +90,7 @@ func (g *Manager) StartEndpointsDetection() {
 					endpoints[ep] = true
 				}
 			} else {
-				klog.Error(err)
+				klog.V(4).Info(err)
 			}
 
 			oldEndpoints := g.balancer.All()
@@ -107,24 +108,22 @@ func (g *Manager) StartEndpointsDetection() {
 					klog.Error(err)
 				} else if state == connectivity.Ready || state == connectivity.Idle {
 					g.balancer.Add(ep)
-				} else {
-					g.balancer.Remove(ep)
 				}
 			}
-			time.Sleep(time.Second)
+			time.Sleep(200 * time.Millisecond)
 		}
 	}()
 }
 
 func (g *Manager) GetGRPCConnection() (*grpc.ClientConn, func(), error) {
 	address, _ := g.balancer.Next(lb.DummyFactor)
 	if address == nil {
-		return nil, nil, errors.Errorf("No available endpoints")
+		return nil, func() {}, errors.Errorf("No available endpoints")
 	}
 
 	conn, teardown, err := g.getGRPCConnection(context.TODO(), address.String(), "", "", true, false, false)
 	if err != nil {
-		return nil, nil, errors.Errorf("error establishing connection to app grpc on address %s: %s", address, err)
+		return nil, teardown, errors.Errorf("error establishing connection to app grpc on address %s: %s", address, err)
 	}
 
 	return conn, func() {