Merge branch 'master' into test-integration-skew

pkg/messaging/v1/invoke_method_request.go

@@ -159,6 +159,12 @@ func (imr *InvokeMethodRequest) WithHTTPExtension(verb string, querystring strin
 // WithCustomHTTPMetadata applies a metadata map to a InvokeMethodRequest.
 func (imr *InvokeMethodRequest) WithCustomHTTPMetadata(md map[string]string) *InvokeMethodRequest {
  for k, v := range md {
+ if strings.EqualFold(k, ContentLengthHeader) {
+ // There is no use of the original payload's content-length because
+ // the entire data is already in the cloud event.
+ continue
+ }
+
  if imr.r.GetMetadata() == nil {
  imr.r.Metadata = make(map[string]*internalv1pb.ListStringValue)
  }

tests/integration/framework/process/daprd/options.go

@@ -23,6 +23,7 @@ import (
 
  "github.com/dapr/dapr/tests/integration/framework/process/exec"
  "github.com/dapr/dapr/tests/integration/framework/process/logline"
+ "github.com/dapr/dapr/tests/integration/framework/socket"
 )
 
 // Option is a function that configures the dapr process.
@@ -280,3 +281,9 @@ func WithControlPlaneTrustDomain(trustDomain string) Option {
  o.controlPlaneTrustDomain = &trustDomain
  }
 }
+
+func WithSocket(t *testing.T, socket *socket.Socket) Option {
+ return WithExecOptions(exec.WithEnvVars(t,
+ "DAPR_COMPONENTS_SOCKETS_FOLDER", socket.Directory(),
+ ))
+}

tests/integration/framework/process/grpc/app/app.go

@@ -41,14 +41,16 @@ func New(t *testing.T, fopts ...Option) *App {
  return &App{
  GRPC: procgrpc.New(t, append(opts.grpcopts, procgrpc.WithRegister(func(s *grpc.Server) {
  srv := &server{
- onInvokeFn: opts.onInvokeFn,
- onTopicEventFn: opts.onTopicEventFn,
- listTopicSubFn: opts.listTopicSubFn,
- listInputBindFn: opts.listInputBindFn,
- onBindingEventFn: opts.onBindingEventFn,
- healthCheckFn: opts.healthCheckFn,
+ onInvokeFn: opts.onInvokeFn,
+ onTopicEventFn: opts.onTopicEventFn,
+ onBulkTopicEventFn: opts.onBulkTopicEventFn,
+ listTopicSubFn: opts.listTopicSubFn,
+ listInputBindFn: opts.listInputBindFn,
+ onBindingEventFn: opts.onBindingEventFn,
+ healthCheckFn: opts.healthCheckFn,
  }
  rtv1.RegisterAppCallbackServer(s, srv)
+ rtv1.RegisterAppCallbackAlphaServer(s, srv)
  rtv1.RegisterAppCallbackHealthCheckServer(s, srv)
  if opts.withRegister != nil {
  opts.withRegister(s)

tests/integration/framework/process/grpc/app/options.go

@@ -26,14 +26,15 @@ import (
 
 // options contains the options for running a GRPC server in integration tests.
 type options struct {
- grpcopts []procgrpc.Option
- withRegister func(s *grpc.Server)
- onTopicEventFn func(context.Context, *rtv1.TopicEventRequest) (*rtv1.TopicEventResponse, error)
- onInvokeFn func(context.Context, *commonv1.InvokeRequest) (*commonv1.InvokeResponse, error)
- listTopicSubFn func(ctx context.Context, in *emptypb.Empty) (*rtv1.ListTopicSubscriptionsResponse, error)
- listInputBindFn func(context.Context, *emptypb.Empty) (*rtv1.ListInputBindingsResponse, error)
- onBindingEventFn func(context.Context, *rtv1.BindingEventRequest) (*rtv1.BindingEventResponse, error)
- healthCheckFn func(context.Context, *emptypb.Empty) (*rtv1.HealthCheckResponse, error)
+ grpcopts []procgrpc.Option
+ withRegister func(s *grpc.Server)
+ onTopicEventFn func(context.Context, *rtv1.TopicEventRequest) (*rtv1.TopicEventResponse, error)
+ onBulkTopicEventFn func(context.Context, *rtv1.TopicEventBulkRequest) (*rtv1.TopicEventBulkResponse, error)
+ onInvokeFn func(context.Context, *commonv1.InvokeRequest) (*commonv1.InvokeResponse, error)
+ listTopicSubFn func(ctx context.Context, in *emptypb.Empty) (*rtv1.ListTopicSubscriptionsResponse, error)
+ listInputBindFn func(context.Context, *emptypb.Empty) (*rtv1.ListInputBindingsResponse, error)
+ onBindingEventFn func(context.Context, *rtv1.BindingEventRequest) (*rtv1.BindingEventResponse, error)
+ healthCheckFn func(context.Context, *emptypb.Empty) (*rtv1.HealthCheckResponse, error)
 }
 
 func WithGRPCOptions(opts ...procgrpc.Option) func(*options) {
@@ -48,6 +49,12 @@ func WithOnTopicEventFn(fn func(context.Context, *rtv1.TopicEventRequest) (*rtv1
  }
 }
 
+func WithOnBulkTopicEventFn(fn func(context.Context, *rtv1.TopicEventBulkRequest) (*rtv1.TopicEventBulkResponse, error)) func(*options) {
+ return func(opts *options) {
+ opts.onBulkTopicEventFn = fn
+ }
+}
+
 func WithOnInvokeFn(fn func(ctx context.Context, in *commonv1.InvokeRequest) (*commonv1.InvokeResponse, error)) func(*options) {
  return func(opts *options) {
  opts.onInvokeFn = fn

tests/integration/framework/process/grpc/app/server.go

@@ -23,12 +23,13 @@ import (
 )
 
 type server struct {
- onInvokeFn func(context.Context, *commonv1.InvokeRequest) (*commonv1.InvokeResponse, error)
- onTopicEventFn func(context.Context, *rtv1.TopicEventRequest) (*rtv1.TopicEventResponse, error)
- listTopicSubFn func(context.Context, *emptypb.Empty) (*rtv1.ListTopicSubscriptionsResponse, error)
- listInputBindFn func(context.Context, *emptypb.Empty) (*rtv1.ListInputBindingsResponse, error)
- onBindingEventFn func(context.Context, *rtv1.BindingEventRequest) (*rtv1.BindingEventResponse, error)
- healthCheckFn func(context.Context, *emptypb.Empty) (*rtv1.HealthCheckResponse, error)
+ onInvokeFn func(context.Context, *commonv1.InvokeRequest) (*commonv1.InvokeResponse, error)
+ onTopicEventFn func(context.Context, *rtv1.TopicEventRequest) (*rtv1.TopicEventResponse, error)
+ onBulkTopicEventFn func(context.Context, *rtv1.TopicEventBulkRequest) (*rtv1.TopicEventBulkResponse, error)
+ listTopicSubFn func(context.Context, *emptypb.Empty) (*rtv1.ListTopicSubscriptionsResponse, error)
+ listInputBindFn func(context.Context, *emptypb.Empty) (*rtv1.ListInputBindingsResponse, error)
+ onBindingEventFn func(context.Context, *rtv1.BindingEventRequest) (*rtv1.BindingEventResponse, error)
+ healthCheckFn func(context.Context, *emptypb.Empty) (*rtv1.HealthCheckResponse, error)
 }
 
 func (s *server) OnInvoke(ctx context.Context, in *commonv1.InvokeRequest) (*commonv1.InvokeResponse, error) {
@@ -66,6 +67,13 @@ func (s *server) OnTopicEvent(ctx context.Context, in *rtv1.TopicEventRequest) (
  return s.onTopicEventFn(ctx, in)
 }
 
+func (s *server) OnBulkTopicEventAlpha1(ctx context.Context, in *rtv1.TopicEventBulkRequest) (*rtv1.TopicEventBulkResponse, error) {
+ if s.onBulkTopicEventFn == nil {
+ return new(rtv1.TopicEventBulkResponse), nil
+ }
+ return s.onBulkTopicEventFn(ctx, in)
+}
+
 func (s *server) HealthCheck(ctx context.Context, e *emptypb.Empty) (*rtv1.HealthCheckResponse, error) {
  if s.healthCheckFn == nil {
  return new(rtv1.HealthCheckResponse), nil

tests/integration/framework/process/grpc/subscriber/options.go

@@ -13,5 +13,21 @@ limitations under the License.
 
 package subscriber
 
+import (
+ "context"
+
+ "google.golang.org/protobuf/types/known/emptypb"
+
+ rtv1 "github.com/dapr/dapr/pkg/proto/runtime/v1"
+)
+
 // options contains the options for running a pubsub subscriber gRPC server app.
-type options struct{}
+type options struct {
+ listTopicSubFn func(ctx context.Context, in *emptypb.Empty) (*rtv1.ListTopicSubscriptionsResponse, error)
+}
+
+func WithListTopicSubscriptions(fn func(ctx context.Context, in *emptypb.Empty) (*rtv1.ListTopicSubscriptionsResponse, error)) func(*options) {
+ return func(opts *options) {
+ opts.listTopicSubFn = fn
+ }
+}

tests/integration/framework/process/grpc/subscriber/subscriber.go

@@ -29,9 +29,10 @@ import (
 type Option func(*options)
 
 type Subscriber struct {
- app *app.App
- inCh chan *rtv1.TopicEventRequest
- closeCh chan struct{}
+ app *app.App
+ inCh chan *rtv1.TopicEventRequest
+ inBulkCh chan *rtv1.TopicEventBulkRequest
+ closeCh chan struct{}
 }
 
 func New(t *testing.T, fopts ...Option) *Subscriber {
@@ -43,12 +44,15 @@ func New(t *testing.T, fopts ...Option) *Subscriber {
  }
 
  inCh := make(chan *rtv1.TopicEventRequest, 100)
+ inBulkCh := make(chan *rtv1.TopicEventBulkRequest, 100)
  closeCh := make(chan struct{})
 
  return &Subscriber{
- inCh: inCh,
- closeCh: closeCh,
+ inCh: inCh,
+ inBulkCh: inBulkCh,
+ closeCh: closeCh,
  app: app.New(t,
+ app.WithListTopicSubscriptions(opts.listTopicSubFn),
  app.WithOnTopicEventFn(func(ctx context.Context, in *rtv1.TopicEventRequest) (*rtv1.TopicEventResponse, error) {
  select {
  case inCh <- in:
@@ -57,6 +61,21 @@ func New(t *testing.T, fopts ...Option) *Subscriber {
  }
  return new(rtv1.TopicEventResponse), nil
  }),
+ app.WithOnBulkTopicEventFn(func(ctx context.Context, in *rtv1.TopicEventBulkRequest) (*rtv1.TopicEventBulkResponse, error) {
+ select {
+ case inBulkCh <- in:
+ case <-ctx.Done():
+ case <-closeCh:
+ }
+ stats := make([]*rtv1.TopicEventBulkResponseEntry, len(in.GetEntries()))
+ for i, e := range in.GetEntries() {
+ stats[i] = &rtv1.TopicEventBulkResponseEntry{
+ EntryId: e.GetEntryId(),
+ Status: rtv1.TopicEventResponse_SUCCESS,
+ }
+ }
+ return &rtv1.TopicEventBulkResponse{Statuses: stats}, nil
+ }),
  ),
  }
 }
@@ -92,11 +111,31 @@ func (s *Subscriber) Receive(t *testing.T, ctx context.Context) *rtv1.TopicEvent
  }
 }
 
+func (s *Subscriber) ReceiveBulk(t *testing.T, ctx context.Context) *rtv1.TopicEventBulkRequest {
+ t.Helper()
+
+ ctx, cancel := context.WithTimeout(ctx, time.Second*3)
+ defer cancel()
+
+ select {
+ case <-ctx.Done():
+ require.Fail(t, "timed out waiting for event response")
+ return nil
+ case in := <-s.inBulkCh:
+ return in
+ }
+}
+
 func (s *Subscriber) AssertEventChanLen(t *testing.T, l int) {
  t.Helper()
  assert.Len(t, s.inCh, l)
 }
 
+func (s *Subscriber) AssertBulkEventChanLen(t *testing.T, l int) {
+ t.Helper()
+ assert.Len(t, s.inBulkCh, l)
+}
+
 func (s *Subscriber) ExpectPublishReceive(t *testing.T, ctx context.Context, daprd *daprd.Daprd, req *rtv1.PublishEventRequest) {
  t.Helper()
  _, err := daprd.GRPCClient(t, ctx).PublishEvent(ctx, req)

tests/integration/framework/process/http/subscriber/subscriber.go

@@ -14,6 +14,7 @@ limitations under the License.
 package subscriber
 
 import (
+ "bytes"
  "context"
  "encoding/json"
  "fmt"
@@ -26,6 +27,7 @@ import (
  "github.com/stretchr/testify/assert"
  "github.com/stretchr/testify/require"
 
+ "github.com/dapr/dapr/pkg/runtime/pubsub"
  "github.com/dapr/dapr/tests/integration/framework/process/daprd"
  "github.com/dapr/dapr/tests/integration/framework/process/http/app"
  "github.com/dapr/dapr/tests/integration/framework/util"
@@ -46,10 +48,24 @@ type PublishRequest struct {
  DataContentType *string
 }
 
+type PublishBulkRequestEntry struct {
+ EntryID string `json:"entryId"`
+ Event string `json:"event"`
+ ContentType string `json:"contentType,omitempty"`
+}
+
+type PublishBulkRequest struct {
+ Daprd *daprd.Daprd
+ PubSubName string
+ Topic string
+ Entries []PublishBulkRequestEntry
+}
+
 type Subscriber struct {
  app *app.App
  client *http.Client
  inCh chan *RouteEvent
+ inBulk chan *pubsub.BulkSubscribeEnvelope
  closeCh chan struct{}
 }
 
@@ -62,9 +78,10 @@ func New(t *testing.T, fopts ...Option) *Subscriber {
  }
 
  inCh := make(chan *RouteEvent, 100)
+ inBulk := make(chan *pubsub.BulkSubscribeEnvelope, 100)
  closeCh := make(chan struct{})
 
- appOpts := make([]app.Option, 0, len(opts.routes)+len(opts.handlerFuncs))
+ appOpts := make([]app.Option, 0, len(opts.routes)+len(opts.bulkRoutes)+len(opts.handlerFuncs))
  for _, route := range opts.routes {
  appOpts = append(appOpts, app.WithHandlerFunc(route, func(w http.ResponseWriter, r *http.Request) {
  var ce event.Event
@@ -76,13 +93,39 @@ func New(t *testing.T, fopts ...Option) *Subscriber {
  }
  }))
  }
+ for _, route := range opts.bulkRoutes {
+ appOpts = append(appOpts, app.WithHandlerFunc(route, func(w http.ResponseWriter, r *http.Request) {
+ var ce pubsub.BulkSubscribeEnvelope
+ require.NoError(t, json.NewDecoder(r.Body).Decode(&ce))
+ select {
+ case inBulk <- &ce:
+ case <-closeCh:
+ case <-r.Context().Done():
+ }
+
+ type statusT struct {
+ EntryID string `json:"entryId"`
+ Status string `json:"status"`
+ }
+ type respT struct {
+ Statuses []statusT `json:"statuses"`
+ }
+
+ var resp respT
+ for _, entry := range ce.Entries {
+ resp.Statuses = append(resp.Statuses, statusT{EntryID: entry.EntryId, Status: "SUCCESS"})
+ }
+ json.NewEncoder(w).Encode(resp)
+ }))
+ }
 
  appOpts = append(appOpts, opts.handlerFuncs...)
 
  return &Subscriber{
  app: app.New(t, appOpts...),
  client: util.HTTPClient(t),
  inCh: inCh,
+ inBulk: inBulk,
  closeCh: closeCh,
  }
 }
@@ -117,6 +160,21 @@ func (s *Subscriber) Receive(t *testing.T, ctx context.Context) *RouteEvent {
  }
 }
 
+func (s *Subscriber) ReceiveBulk(t *testing.T, ctx context.Context) *pubsub.BulkSubscribeEnvelope {
+ t.Helper()
+
+ ctx, cancel := context.WithTimeout(ctx, time.Second)
+ defer cancel()
+
+ select {
+ case <-ctx.Done():
+ require.Fail(t, "timed out waiting for event response")
+ return nil
+ case in := <-s.inBulk:
+ return in
+ }
+}
+
 func (s *Subscriber) AssertEventChanLen(t *testing.T, l int) {
  t.Helper()
  assert.Len(t, s.inCh, l)
@@ -151,6 +209,13 @@ func (s *Subscriber) Publish(t *testing.T, ctx context.Context, req PublishReque
  require.Equal(t, http.StatusNoContent, resp.StatusCode)
 }
 
+func (s *Subscriber) PublishBulk(t *testing.T, ctx context.Context, req PublishBulkRequest) {
+ t.Helper()
+ //nolint:bodyclose
+ resp := s.publishBulk(t, ctx, req)
+ require.Equal(t, http.StatusNoContent, resp.StatusCode)
+}
+
 func (s *Subscriber) publish(t *testing.T, ctx context.Context, req PublishRequest) *http.Response {
  t.Helper()
  reqURL := fmt.Sprintf("http://%s/v1.0/publish/%s/%s", req.Daprd.HTTPAddress(), req.PubSubName, req.Topic)
@@ -164,3 +229,18 @@ func (s *Subscriber) publish(t *testing.T, ctx context.Context, req PublishReque
  require.NoError(t, resp.Body.Close())
  return resp
 }
+
+func (s *Subscriber) publishBulk(t *testing.T, ctx context.Context, req PublishBulkRequest) *http.Response {
+ t.Helper()
+
+ payload, err := json.Marshal(req.Entries)
+ require.NoError(t, err)
+ reqURL := fmt.Sprintf("http://%s/v1.0-alpha1/publish/bulk/%s/%s", req.Daprd.HTTPAddress(), req.PubSubName, req.Topic)
+ hreq, err := http.NewRequestWithContext(ctx, http.MethodPost, reqURL, bytes.NewReader(payload))
+ require.NoError(t, err)
+ hreq.Header.Add("Content-Type", "application/json")
+ resp, err := s.client.Do(hreq)
+ require.NoError(t, err)
+ require.NoError(t, resp.Body.Close())
+ return resp
+}