You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by pi...@apache.org on 2025/07/30 09:05:44 UTC
(iggy) branch master updated: feat(go): add godog go bdd test according to scenarios. (#2021)
This is an automated email from the ASF dual-hosted git repository.
piotr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 8cfd671c feat(go): add godog go bdd test according to scenarios. (#2021)
8cfd671c is described below
commit 8cfd671ca9f67026875b461a18349bcd339328e5
Author: Chengxi <ch...@gmail.com>
AuthorDate: Wed Jul 30 05:05:40 2025 -0400
feat(go): add godog go bdd test according to scenarios. (#2021)
This PR handles the issue #1986
1. add a real bdd test for golang sdk.
2. update the Dockerfile and docker-compose
3. fix a small bug in iggy message header deserialization
---
bdd/docker-compose.yml | 2 +-
bdd/go/Dockerfile | 2 +-
bdd/go/go.mod | 8 +
bdd/go/go.sum | 54 ++++-
bdd/go/tests/basic_messaging_test.go | 353 +++++++++++++++++++++++++++++++++
foreign/go/contracts/message_header.go | 6 +-
6 files changed, 416 insertions(+), 9 deletions(-)
diff --git a/bdd/docker-compose.yml b/bdd/docker-compose.yml
index 1fc1b488..fb2a6c1a 100644
--- a/bdd/docker-compose.yml
+++ b/bdd/docker-compose.yml
@@ -71,7 +71,7 @@ services:
- IGGY_TCP_ADDRESS=iggy-server:8090
volumes:
- ./scenarios/basic_messaging.feature:/app/features/basic_messaging.feature
- command: [ "ginkgo", "-v", "-r" ]
+ command: [ "go", "test", "-v", "./..." ]
node-bdd:
build:
diff --git a/bdd/go/Dockerfile b/bdd/go/Dockerfile
index 658f6f44..0d933b74 100644
--- a/bdd/go/Dockerfile
+++ b/bdd/go/Dockerfile
@@ -27,4 +27,4 @@ RUN go install github.com/onsi/ginkgo/v2/ginkgo@latest
WORKDIR /app/bdd/go
# Default command
-CMD ["ginkgo", "-v", "-r"]
+CMD ["go", "test", "-v", "./..."]
diff --git a/bdd/go/go.mod b/bdd/go/go.mod
index 7570f031..b05137a8 100644
--- a/bdd/go/go.mod
+++ b/bdd/go/go.mod
@@ -6,17 +6,25 @@ replace github.com/apache/iggy/foreign/go => ../../foreign/go
require (
github.com/apache/iggy/foreign/go v0.0.0-00010101000000-000000000000
+ github.com/cucumber/godog v0.15.1
github.com/google/uuid v1.6.0
github.com/onsi/ginkgo/v2 v2.23.4
github.com/onsi/gomega v1.37.0
)
require (
+ github.com/cucumber/gherkin/go/v26 v26.2.0 // indirect
+ github.com/cucumber/messages/go/v21 v21.0.1 // indirect
github.com/go-logr/logr v1.4.2 // indirect
github.com/go-task/slim-sprig/v3 v3.0.0 // indirect
+ github.com/gofrs/uuid v4.4.0+incompatible // indirect
github.com/google/go-cmp v0.7.0 // indirect
github.com/google/pprof v0.0.0-20250403155104-27863c87afa6 // indirect
+ github.com/hashicorp/go-immutable-radix v1.3.1 // indirect
+ github.com/hashicorp/go-memdb v1.3.5 // indirect
+ github.com/hashicorp/golang-lru v1.0.2 // indirect
github.com/klauspost/compress v1.17.11 // indirect
+ github.com/spf13/pflag v1.0.7 // indirect
go.uber.org/automaxprocs v1.6.0 // indirect
golang.org/x/net v0.38.0 // indirect
golang.org/x/sys v0.32.0 // indirect
diff --git a/bdd/go/go.sum b/bdd/go/go.sum
index f31ed10d..86932a34 100644
--- a/bdd/go/go.sum
+++ b/bdd/go/go.sum
@@ -1,19 +1,52 @@
+github.com/cpuguy83/go-md2man/v2 v2.0.2/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o=
+github.com/cucumber/gherkin/go/v26 v26.2.0 h1:EgIjePLWiPeslwIWmNQ3XHcypPsWAHoMCz/YEBKP4GI=
+github.com/cucumber/gherkin/go/v26 v26.2.0/go.mod h1:t2GAPnB8maCT4lkHL99BDCVNzCh1d7dBhCLt150Nr/0=
+github.com/cucumber/godog v0.15.0/go.mod h1:FX3rzIDybWABU4kuIXLZ/qtqEe1Ac5RdXmqvACJOces=
+github.com/cucumber/godog v0.15.1 h1:rb/6oHDdvVZKS66hrhpjFQFHjthFSrQBCOI1LwshNTI=
+github.com/cucumber/godog v0.15.1/go.mod h1:qju+SQDewOljHuq9NSM66s0xEhogx0q30flfxL4WUk8=
+github.com/cucumber/messages/go/v21 v21.0.1 h1:wzA0LxwjlWQYZd32VTlAVDTkW6inOFmSM+RuOwHZiMI=
+github.com/cucumber/messages/go/v21 v21.0.1/go.mod h1:zheH/2HS9JLVFukdrsPWoPdmUtmYQAQPLk7w5vWsk5s=
+github.com/cucumber/messages/go/v22 v22.0.0/go.mod h1:aZipXTKc0JnjCsXrJnuZpWhtay93k7Rn3Dee7iyPJjs=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/go-logr/logr v1.4.2 h1:6pFjapn8bFcIbiKo3XT4j/BhANplGihG6tvd+8rYgrY=
github.com/go-logr/logr v1.4.2/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
github.com/go-task/slim-sprig/v3 v3.0.0 h1:sUs3vkvUymDpBKi3qH1YSqBQk9+9D/8M2mN1vB6EwHI=
github.com/go-task/slim-sprig/v3 v3.0.0/go.mod h1:W848ghGpv3Qj3dhTPRyJypKRiqCdHZiAzKg9hl15HA8=
+github.com/gofrs/uuid v4.2.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM=
+github.com/gofrs/uuid v4.3.1+incompatible h1:0/KbAdpx3UXAx1kEOWHJeOkpbgRFGHVgv+CFIY7dBJI=
+github.com/gofrs/uuid v4.3.1+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM=
+github.com/gofrs/uuid v4.4.0+incompatible h1:3qXRTX8/NbyulANqlc0lchS1gqAVxRgsuW1YrTJupqA=
+github.com/gofrs/uuid v4.4.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM=
github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
github.com/google/pprof v0.0.0-20250403155104-27863c87afa6 h1:BHT72Gu3keYf3ZEu2J0b1vyeLSOYI8bm5wbJM/8yDe8=
github.com/google/pprof v0.0.0-20250403155104-27863c87afa6/go.mod h1:boTsfXsheKC2y+lKOCMpSfarhxDeIzfZG1jqGcPl3cA=
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/hashicorp/go-immutable-radix v1.3.0/go.mod h1:0y9vanUI8NX6FsYoO3zeMjhV/C5i9g4Q3DwcSNZ4P60=
+github.com/hashicorp/go-immutable-radix v1.3.1 h1:DKHmCUm2hRBK510BaiZlwvpD40f8bJFeZnpfm2KLowc=
+github.com/hashicorp/go-immutable-radix v1.3.1/go.mod h1:0y9vanUI8NX6FsYoO3zeMjhV/C5i9g4Q3DwcSNZ4P60=
+github.com/hashicorp/go-memdb v1.3.4 h1:XSL3NR682X/cVk2IeV0d70N4DZ9ljI885xAEU8IoK3c=
+github.com/hashicorp/go-memdb v1.3.4/go.mod h1:uBTr1oQbtuMgd1SSGoR8YV27eT3sBHbYiNm53bMpgSg=
+github.com/hashicorp/go-memdb v1.3.5 h1:b3taDMxCBCBVgyRrS1AZVHO14ubMYZB++QpNhBg+Nyo=
+github.com/hashicorp/go-memdb v1.3.5/go.mod h1:8IVKKBkVe+fxFgdFOYxzQQNjz+sWCyHCdIC/+5+Vy1Y=
+github.com/hashicorp/go-uuid v1.0.0/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
+github.com/hashicorp/go-uuid v1.0.2 h1:cfejS+Tpcp13yd5nYHWDI6qVCny6wyX2Mt5SGur2IGE=
+github.com/hashicorp/go-uuid v1.0.2/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
+github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
+github.com/hashicorp/golang-lru v0.5.4 h1:YDjusn29QI/Das2iO9M0BHnIbxPeyuCHsjMW+lJfyTc=
+github.com/hashicorp/golang-lru v0.5.4/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
+github.com/hashicorp/golang-lru v1.0.2 h1:dV3g9Z/unq5DpblPpw+Oqcv4dU/1omnb4Ok8iPY6p1c=
+github.com/hashicorp/golang-lru v1.0.2/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
+github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
github.com/klauspost/compress v1.17.11 h1:In6xLpyWOi1+C7tXUUWv2ot1QvBjxevKAaI6IXrJmUc=
github.com/klauspost/compress v1.17.11/go.mod h1:pMDklpSncoRMuLFrf1W9Ss9KT+0rH90U12bZKk7uwG0=
-github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
-github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
+github.com/kr/pretty v0.2.1 h1:Fmg33tUaq4/8ym9TJN1x7sLJnHVwhP33CNkpYV/7rwI=
+github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
+github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
+github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
github.com/onsi/ginkgo/v2 v2.23.4 h1:ktYTpKJAVZnDT4VjxSbiBenUjmlL/5QkBEocaWXiQus=
@@ -24,6 +57,18 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/prashantv/gostub v1.1.0 h1:BTyx3RfQjRHnUWaGF9oQos79AlQ5k8WNktv7VGvVH4g=
github.com/prashantv/gostub v1.1.0/go.mod h1:A5zLQHz7ieHGG7is6LLXLz7I8+3LZzsrV0P1IAHhP5U=
+github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
+github.com/spf13/cobra v1.7.0/go.mod h1:uLxZILRyS/50WlhOIKD7W6V5bgeIt+4sICxh6uRMrb0=
+github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
+github.com/spf13/pflag v1.0.7 h1:vN6T9TfwStFPFM5XzjsvmzZkLuaLX+HS+0SeFLRgU6M=
+github.com/spf13/pflag v1.0.7/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
+github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
+github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
+github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
+github.com/stretchr/testify v1.8.2/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
go.uber.org/automaxprocs v1.6.0 h1:O3y2/QNTOdbF+e/dpXNNW7Rx2hZ4sTIPyybbxyNqTUs=
@@ -39,7 +84,8 @@ golang.org/x/tools v0.31.0/go.mod h1:naFTU+Cev749tSJRXJlna0T3WxKvb1kWEx15xA4SdmQ
google.golang.org/protobuf v1.36.5 h1:tPhr+woSbjfYvY6/GPufUoYizxw1cF/yFoxJ2fmpwlM=
google.golang.org/protobuf v1.36.5/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY=
-gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
diff --git a/bdd/go/tests/basic_messaging_test.go b/bdd/go/tests/basic_messaging_test.go
new file mode 100644
index 00000000..7c4232a5
--- /dev/null
+++ b/bdd/go/tests/basic_messaging_test.go
@@ -0,0 +1,353 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package tests
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ iggcon "github.com/apache/iggy/foreign/go/contracts"
+ "github.com/apache/iggy/foreign/go/iggycli"
+ "github.com/apache/iggy/foreign/go/tcp"
+ "github.com/cucumber/godog"
+ "github.com/google/uuid"
+ "os"
+ "testing"
+)
+
+type basicMessagingCtxKey struct{}
+
+type basicMessagingCtx struct {
+ serverAddr *string
+ client iggycli.Client
+ lastSentMessage *iggcon.IggyMessage
+ lastPollMessages *iggcon.PolledMessage
+ lastStreamID *uint32
+ lastStreamName *string
+ lastTopicID *uint32
+ lastTopicName *string
+ lastTopicPartitions *uint32
+}
+
+func getBasicMessagingCtx(ctx context.Context) *basicMessagingCtx {
+ return ctx.Value(basicMessagingCtxKey{}).(*basicMessagingCtx)
+}
+
+func givenRunningServer(ctx context.Context) error {
+ c := getBasicMessagingCtx(ctx)
+ addr := os.Getenv("IGGY_TCP_ADDRESS")
+ if addr == "" {
+ addr = "127.0.0.1:8090"
+ }
+ c.serverAddr = &addr
+ return nil
+}
+func givenAuthenticationAsRoot(ctx context.Context) error {
+ c := getBasicMessagingCtx(ctx)
+ serverAddr := *c.serverAddr
+
+ client, err := iggycli.NewIggyClient(
+ iggycli.WithTcp(
+ tcp.WithServerAddress(serverAddr),
+ ),
+ )
+ if err != nil {
+ return fmt.Errorf("error creating client: %w", err)
+ }
+
+ if err = client.Ping(); err != nil {
+ return fmt.Errorf("error pinging client: %w", err)
+ }
+
+ if _, err = client.LoginUser("iggy", "iggy"); err != nil {
+ return fmt.Errorf("error logging in: %v", err)
+ }
+
+ c.client = client
+ return nil
+}
+
+func whenSendMessages(
+ ctx context.Context,
+ messagesCount uint32,
+ streamID uint32,
+ topicID uint32,
+ partitionID uint32) error {
+ c := getBasicMessagingCtx(ctx)
+ messages, err := createTestMessages(messagesCount)
+ if err != nil {
+ return fmt.Errorf("error creating test messages: %w", err)
+ }
+
+ streamIdentifier, _ := iggcon.NewIdentifier(streamID)
+ topicIdentifier, _ := iggcon.NewIdentifier(topicID)
+ partitioning := iggcon.PartitionId(partitionID)
+ if err = c.client.SendMessages(streamIdentifier, topicIdentifier, partitioning, messages); err != nil {
+ return fmt.Errorf("failed to sending messages: %w", err)
+ }
+
+ c.lastSentMessage = &messages[len(messages)-1]
+ return nil
+}
+
+func createTestMessages(count uint32) ([]iggcon.IggyMessage, error) {
+ messages := make([]iggcon.IggyMessage, 0, count)
+ for i := 0; uint32(i) < count; i++ {
+ id := uuid.New()
+ payload := []byte(fmt.Sprintf("test message %d", i))
+ message, err := iggcon.NewIggyMessage(payload, iggcon.WithID(id))
+ if err != nil {
+ return nil, fmt.Errorf("failed to create message: %w", err)
+ }
+ messages = append(messages, message)
+ }
+ return messages, nil
+}
+
+func whenPollMessages(
+ ctx context.Context,
+ streamID uint32,
+ topicID uint32,
+ partitionID uint32,
+ startOffset uint64) error {
+ c := getBasicMessagingCtx(ctx)
+ consumer := iggcon.DefaultConsumer()
+ streamIdentifier, _ := iggcon.NewIdentifier(streamID)
+ topicIdentifier, _ := iggcon.NewIdentifier(topicID)
+ uint32PartitionID := partitionID
+ polledMessages, err := c.client.PollMessages(
+ streamIdentifier,
+ topicIdentifier,
+ consumer,
+ iggcon.OffsetPollingStrategy(startOffset),
+ 100,
+ false,
+ &uint32PartitionID,
+ )
+ if err != nil {
+ return fmt.Errorf("failed to poll messages: %w", err)
+ }
+ c.lastPollMessages = polledMessages
+ return nil
+}
+
+func thenMessageSentSuccessfully(_ context.Context) error {
+ return nil
+}
+
+func thenShouldReceiveMessages(ctx context.Context, expectedCount uint32) error {
+ polledMessages := getBasicMessagingCtx(ctx).lastPollMessages
+ if uint32(len(polledMessages.Messages)) != expectedCount {
+ return fmt.Errorf("expected %d messages, but there is %d", expectedCount, len(polledMessages.Messages))
+ }
+ return nil
+}
+
+func thenMessagesHaveSequentialOffsets(
+ ctx context.Context,
+ startOffset uint64,
+ endOffset uint64) error {
+ polledMessages := getBasicMessagingCtx(ctx).lastPollMessages
+ for i, m := range polledMessages.Messages {
+ expectedOffset := startOffset + uint64(i)
+ if expectedOffset != m.Header.Offset {
+ return fmt.Errorf("message at index %d should have offset %d", i, expectedOffset)
+ }
+ }
+ lastMessage := polledMessages.Messages[len(polledMessages.Messages)-1]
+ if lastMessage.Header.Offset != endOffset {
+ return fmt.Errorf("last message should have offset %d", endOffset)
+ }
+ return nil
+}
+
+func thenMessagesHaveExpectedPayload(ctx context.Context) error {
+ polledMessages := getBasicMessagingCtx(ctx).lastPollMessages
+ for i, m := range polledMessages.Messages {
+ expectedPayload := fmt.Sprintf("test message %d", i)
+ if expectedPayload != string(m.Payload) {
+ return fmt.Errorf("message at offset %d should have payload '%s'", i, expectedPayload)
+ }
+ }
+ return nil
+}
+
+func thenLastPolledMessageMatchesSent(ctx context.Context) error {
+ c := getBasicMessagingCtx(ctx)
+ polledMessages := c.lastPollMessages
+ sentMessage := c.lastSentMessage
+ if len(polledMessages.Messages) == 0 {
+ return errors.New("should have at least one polled message")
+ }
+
+ lastPolled := polledMessages.Messages[len(polledMessages.Messages)-1]
+
+ if lastPolled.Header.Id != sentMessage.Header.Id {
+ return fmt.Errorf("message IDs should match: sent %d, polled %d", sentMessage.Header.Id, lastPolled.Header.Id)
+ }
+
+ if string(lastPolled.Payload) != string(sentMessage.Payload) {
+ return fmt.Errorf("message payload should match: sent %s, polled %s", sentMessage.Payload, lastPolled.Header.Id)
+ }
+ return nil
+}
+
+func givenNoStreams(ctx context.Context) error {
+ client := getBasicMessagingCtx(ctx).client
+ streams, err := client.GetStreams()
+ if err != nil {
+ return fmt.Errorf("failed to get streams: %w", err)
+ }
+
+ if len(streams) != 0 {
+ return errors.New("system should have no stream initially")
+ }
+
+ return err
+}
+
+func whenCreateStream(ctx context.Context, streamID uint32, streamName string) error {
+ c := getBasicMessagingCtx(ctx)
+ stream, err := c.client.CreateStream(streamName, &streamID)
+ if err != nil {
+ return fmt.Errorf("failed to create stream: %w", err)
+ }
+ c.lastStreamID = &stream.Id
+ c.lastStreamName = &stream.Name
+ return nil
+}
+
+func thenStreamCreatedSuccessfully(ctx context.Context) error {
+ if getBasicMessagingCtx(ctx).lastStreamID == nil {
+ return errors.New("stream should have been created")
+ }
+ return nil
+}
+
+func thenStreamHasIDAndName(
+ ctx context.Context,
+ expectedID uint32,
+ expectedName string) error {
+ c := getBasicMessagingCtx(ctx)
+ streamID := *c.lastStreamID
+ streamName := *c.lastStreamName
+ if streamID != expectedID {
+ return fmt.Errorf("expected stream ID %d, got %d", expectedID, streamID)
+ }
+ if streamName != expectedName {
+ return fmt.Errorf("expected stream name %s, got %s", expectedName, streamName)
+ }
+ return nil
+}
+
+func whenCreateTopic(
+ ctx context.Context,
+ topicID uint32,
+ topicName string,
+ streamID uint32,
+ partitionsCount uint32) error {
+ c := getBasicMessagingCtx(ctx)
+ streamIdentifier, _ := iggcon.NewIdentifier(streamID)
+ uint32TopicID := topicID
+ topic, err := c.client.CreateTopic(
+ streamIdentifier,
+ topicName,
+ partitionsCount,
+ iggcon.CompressionAlgorithmNone,
+ iggcon.IggyExpiryNeverExpire,
+ 0,
+ nil,
+ &uint32TopicID,
+ )
+ if err != nil {
+ return fmt.Errorf("failed to create topic: %w", err)
+ }
+
+ c.lastTopicID = &topic.Id
+ c.lastTopicName = &topic.Name
+ c.lastTopicPartitions = &topic.PartitionsCount
+
+ return nil
+}
+
+func thenTopicCreatedSuccessfully(ctx context.Context) error {
+ if getBasicMessagingCtx(ctx).lastTopicID == nil {
+ return errors.New("topic should have been created")
+ }
+ return nil
+}
+func thenTopicHasIDAndName(
+ ctx context.Context,
+ expectedID uint32,
+ expectedName string) error {
+ c := getBasicMessagingCtx(ctx)
+ topicID := *c.lastTopicID
+ topicName := *c.lastTopicName
+ if topicID != expectedID {
+ return fmt.Errorf("expected topic ID %d, got %d", expectedID, topicID)
+ }
+ if topicName != expectedName {
+ return fmt.Errorf("expected topic name %s, got %s", expectedName, topicName)
+ }
+ return nil
+}
+
+func thenTopicsHasPartitions(ctx context.Context, expectedTopicPartitions uint32) error {
+ topicPartitions := *getBasicMessagingCtx(ctx).lastTopicPartitions
+ if topicPartitions != expectedTopicPartitions {
+ return errors.New("topic should have expected number of partitions")
+ }
+ return nil
+}
+
+func initScenarios(sc *godog.ScenarioContext) {
+ sc.Before(func(ctx context.Context, sc *godog.Scenario) (context.Context, error) {
+ return context.WithValue(context.Background(), basicMessagingCtxKey{}, &basicMessagingCtx{}), nil
+ })
+ sc.Step(`I have a running Iggy server`, givenRunningServer)
+ sc.Step(`I am authenticated as the root user`, givenAuthenticationAsRoot)
+ sc.Step(`^I send (\d+) messages to stream (\d+), topic (\d+), partition (\d+)$`, whenSendMessages)
+ sc.Step(`^I poll messages from stream (\d+), topic (\d+), partition (\d+) starting from offset (\d+)$`, whenPollMessages)
+ sc.Step(`all messages should be sent successfully`, thenMessageSentSuccessfully)
+ sc.Step(`^I should receive (\d+) messages$`, thenShouldReceiveMessages)
+ sc.Step(`^the messages should have sequential offsets from (\d+) to (\d+)$`, thenMessagesHaveSequentialOffsets)
+ sc.Step(`each message should have the expected payload content`, thenMessagesHaveExpectedPayload)
+ sc.Step(`the last polled message should match the last sent message`, thenLastPolledMessageMatchesSent)
+ sc.Step(`^the stream should have ID (\d+) and name (.+)$`, thenStreamHasIDAndName)
+ sc.Step(`the stream should be created successfully`, thenStreamCreatedSuccessfully)
+ sc.Step(`^I create a stream with ID (\d+) and name (.+)$`, whenCreateStream)
+ sc.Step(`I have no streams in the system`, givenNoStreams)
+ sc.Step(`^I create a topic with ID (\d+) and name (.+) in stream (\d+) with (\d+) partitions$`, whenCreateTopic)
+ sc.Step(`the topic should be created successfully`, thenTopicCreatedSuccessfully)
+ sc.Step(`^the topic should have ID (\d+) and name (.+)$`, thenTopicHasIDAndName)
+ sc.Step(`^the topic should have (\d+) partitions$`, thenTopicsHasPartitions)
+}
+
+func TestFeatures(t *testing.T) {
+ suite := godog.TestSuite{
+ ScenarioInitializer: initScenarios,
+ Options: &godog.Options{
+ Format: "pretty",
+ Paths: []string{"../../scenarios/basic_messaging.feature"},
+ TestingT: t,
+ },
+ }
+ if suite.Run() != 0 {
+ t.Fatal("failing feature tests")
+ }
+}
diff --git a/foreign/go/contracts/message_header.go b/foreign/go/contracts/message_header.go
index 30b630f3..88afa0c9 100644
--- a/foreign/go/contracts/message_header.go
+++ b/foreign/go/contracts/message_header.go
@@ -53,9 +53,9 @@ func MessageHeaderFromBytes(data []byte) (*MessageHeader, error) {
}
checksum := binary.LittleEndian.Uint64(data[0:8])
id := data[8:24]
- timestamp := binary.LittleEndian.Uint64(data[24:32])
- originTimestamp := binary.LittleEndian.Uint64(data[32:40])
- offset := binary.LittleEndian.Uint64(data[40:48])
+ offset := binary.LittleEndian.Uint64(data[24:32])
+ timestamp := binary.LittleEndian.Uint64(data[32:40])
+ originTimestamp := binary.LittleEndian.Uint64(data[40:48])
userHeaderLength := binary.LittleEndian.Uint32(data[48:52])
payloadLength := binary.LittleEndian.Uint32(data[52:56])