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/04 09:05:21 UTC
(iggy) branch master updated: refactor(sdk): refactor Go SDK client (#1916)
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 82e35db9 refactor(sdk): refactor Go SDK client (#1916)
82e35db9 is described below
commit 82e35db9e7346f321256e1a0a243abbe9be593b0
Author: Chengxi <ch...@gmail.com>
AuthorDate: Fri Jul 4 05:05:18 2025 -0400
refactor(sdk): refactor Go SDK client (#1916)
- Refactored the Go SDK client creation logic using the option pattern.
- Updated the client interface and TCP client methods to align closely with the Rust SDK:
- Replaced struct-based parameters with explicit argument lists.
- Used pointers to represent optional arguments.
- Edited some return type to align with rust SDK.
- Changed all related tests to reflect the updated interface and logic.
---
bdd/go/tests/tcp_test/client_feature_get_all.go | 2 +-
bdd/go/tests/tcp_test/consumers_feature_create.go | 100 ++++----
bdd/go/tests/tcp_test/consumers_feature_delete.go | 52 ++---
bdd/go/tests/tcp_test/consumers_feature_get_all.go | 2 +-
.../tests/tcp_test/consumers_feature_get_by_id.go | 12 +-
bdd/go/tests/tcp_test/consumers_feature_join.go | 51 ++--
bdd/go/tests/tcp_test/consumers_feature_leave.go | 52 ++---
bdd/go/tests/tcp_test/consumers_steps.go | 60 ++---
bdd/go/tests/tcp_test/messages_feature_send.go | 80 +++----
bdd/go/tests/tcp_test/messages_steps.go | 22 +-
bdd/go/tests/tcp_test/partitions_feature_create.go | 51 ++--
bdd/go/tests/tcp_test/partitions_feature_delete.go | 51 ++--
bdd/go/tests/tcp_test/partitions_steps.go | 6 +-
bdd/go/tests/tcp_test/pat_feature_create.go | 20 +-
bdd/go/tests/tcp_test/pat_feature_delete.go | 11 +-
bdd/go/tests/tcp_test/pat_feature_get_all.go | 6 +-
bdd/go/tests/tcp_test/pat_steps.go | 27 +--
bdd/go/tests/tcp_test/ping_feature.go | 2 +-
bdd/go/tests/tcp_test/session_feature_login.go | 28 +--
bdd/go/tests/tcp_test/session_feature_logout.go | 6 +-
bdd/go/tests/tcp_test/stream_feature_create.go | 60 ++---
bdd/go/tests/tcp_test/stream_feature_delete.go | 2 +-
bdd/go/tests/tcp_test/stream_feature_get_all.go | 2 +-
bdd/go/tests/tcp_test/stream_feature_get_by_id.go | 6 +-
bdd/go/tests/tcp_test/stream_feature_update.go | 27 +--
bdd/go/tests/tcp_test/stream_steps.go | 35 ++-
bdd/go/tests/tcp_test/test_helpers.go | 31 ++-
bdd/go/tests/tcp_test/topic_feature_create.go | 138 +++++------
bdd/go/tests/tcp_test/topic_feature_delete.go | 2 +-
bdd/go/tests/tcp_test/topic_feature_get_all.go | 2 +-
bdd/go/tests/tcp_test/topic_feature_get_by_id.go | 8 +-
bdd/go/tests/tcp_test/topic_feature_update.go | 102 ++++----
bdd/go/tests/tcp_test/topic_steps.go | 51 ++--
bdd/go/tests/tcp_test/users_feature_create.go | 57 +++--
bdd/go/tests/tcp_test/users_feature_delete.go | 2 +-
bdd/go/tests/tcp_test/users_feature_get_all.go | 2 +-
bdd/go/tests/tcp_test/users_feature_get_by_id.go | 2 +-
bdd/go/tests/tcp_test/users_feature_password.go | 41 ++--
bdd/go/tests/tcp_test/users_feature_permissions.go | 26 +--
bdd/go/tests/tcp_test/users_feature_update.go | 26 ++-
bdd/go/tests/tcp_test/users_steps.go | 46 ++--
.../go/benchmarks/send_messages_benchmark_test.go | 83 ++++---
.../binary_request_serializer.go | 49 ++--
.../binary_response_deserializer.go | 177 +++++++-------
.../create_stream_serializer.go | 11 +-
.../create_stream_serializer_test.go | 8 +-
.../create_topic_serializer.go | 21 +-
.../fetch_messages_request_serializer.go | 19 +-
.../fetch_messages_request_serializer_test.go | 21 +-
.../log_in_request_serializer.go | 9 +-
.../send_messages_request_serializer.go | 5 +-
.../send_messages_request_serializer_test.go | 14 +-
.../update_stream_serializer.go | 3 +-
.../update_stream_serializer_test.go | 9 +-
.../update_topic_serializer.go | 14 +-
.../update_topic_serializer_test.go | 12 +-
foreign/go/cli/cli.go | 80 +++----
foreign/go/contracts/access_tokens.go | 8 +-
foreign/go/contracts/client.go | 18 +-
foreign/go/contracts/config.go | 13 --
foreign/go/contracts/consumer.go | 4 +-
foreign/go/contracts/consumer_groups.go | 15 +-
foreign/go/contracts/login.go | 9 +-
foreign/go/contracts/messages.go | 4 +-
foreign/go/contracts/offets.go | 10 +-
foreign/go/contracts/partitions.go | 6 +-
foreign/go/contracts/stream.go | 21 +-
foreign/go/contracts/topics.go | 28 ++-
foreign/go/contracts/users.go | 18 +-
foreign/go/errors/constants.go | 8 +
foreign/go/iggycli/client.go | 260 +++++++++++++++++++++
foreign/go/iggycli/iggy_client.go | 71 ++++++
foreign/go/message_stream.go | 72 ------
foreign/go/message_stream_factory.go | 55 -----
foreign/go/samples/consumer/consumer.go | 77 +++---
foreign/go/samples/producer/producer.go | 67 +++---
foreign/go/tcp/tcp_access_token_managament.go | 15 +-
foreign/go/tcp/tcp_clients_managament.go | 4 +-
foreign/go/tcp/tcp_consumer_group_managament.go | 40 ++--
foreign/go/tcp/tcp_core.go | 68 ++++--
foreign/go/tcp/tcp_messaging.go | 36 ++-
foreign/go/tcp/tcp_offset_managament.go | 19 +-
foreign/go/tcp/tcp_partition_managament.go | 16 +-
foreign/go/tcp/tcp_session_managament.go | 15 +-
foreign/go/tcp/tcp_stream_managament.go | 33 +--
foreign/go/tcp/tcp_topic_managament.go | 61 ++++-
foreign/go/tcp/tcp_user_managament.go | 47 +++-
87 files changed, 1623 insertions(+), 1339 deletions(-)
diff --git a/bdd/go/tests/tcp_test/client_feature_get_all.go b/bdd/go/tests/tcp_test/client_feature_get_all.go
index 7cb799cf..c1cd2a1b 100644
--- a/bdd/go/tests/tcp_test/client_feature_get_all.go
+++ b/bdd/go/tests/tcp_test/client_feature_get_all.go
@@ -41,7 +41,7 @@ var _ = Describe("GET ALL CLIENT FEATURE:", func() {
When("user is not logged in", func() {
Context("and tries get all clients", func() {
- client := createConnection()
+ client := createClient()
clients, err := client.GetClients()
itShouldReturnUnauthenticatedError(err)
diff --git a/bdd/go/tests/tcp_test/consumers_feature_create.go b/bdd/go/tests/tcp_test/consumers_feature_create.go
index c4df1cfc..6482266e 100644
--- a/bdd/go/tests/tcp_test/consumers_feature_create.go
+++ b/bdd/go/tests/tcp_test/consumers_feature_create.go
@@ -30,28 +30,27 @@ var _ = Describe("CREATE CONSUMER GROUP:", func() {
streamId, _ := successfullyCreateStream(prefix, client)
defer deleteStreamAfterTests(streamId, client)
topicId, _ := successfullyCreateTopic(streamId, client)
-
- request := iggcon.CreateConsumerGroupRequest{
- StreamId: iggcon.NewIdentifier(streamId),
- TopicId: iggcon.NewIdentifier(topicId),
- ConsumerGroupId: int(createRandomUInt32()),
- Name: createRandomString(16),
- }
- err := client.CreateConsumerGroup(request)
+ groupId := createRandomUInt32()
+ name := createRandomString(16)
+ _, err := client.CreateConsumerGroup(
+ iggcon.NewIdentifier(streamId),
+ iggcon.NewIdentifier(topicId),
+ name,
+ &groupId,
+ )
itShouldNotReturnError(err)
- itShouldSuccessfullyCreateConsumer(streamId, topicId, request.ConsumerGroupId, request.Name, client)
+ itShouldSuccessfullyCreateConsumer(streamId, topicId, int(groupId), name, client)
})
Context("and tries to create consumer group for a non existing stream", func() {
client := createAuthorizedConnection()
- request := iggcon.CreateConsumerGroupRequest{
- StreamId: iggcon.NewIdentifier(int(createRandomUInt32())),
- TopicId: iggcon.NewIdentifier(int(createRandomUInt32())),
- ConsumerGroupId: int(createRandomUInt32()),
- Name: createRandomString(16),
- }
- err := client.CreateConsumerGroup(request)
+ groupId := createRandomUInt32()
+ _, err := client.CreateConsumerGroup(
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ createRandomString(16),
+ &groupId)
itShouldReturnSpecificError(err, "stream_id_not_found")
})
@@ -60,13 +59,13 @@ var _ = Describe("CREATE CONSUMER GROUP:", func() {
client := createAuthorizedConnection()
streamId, _ := successfullyCreateStream(prefix, client)
defer deleteStreamAfterTests(streamId, client)
- request := iggcon.CreateConsumerGroupRequest{
- StreamId: iggcon.NewIdentifier(streamId),
- TopicId: iggcon.NewIdentifier(int(createRandomUInt32())),
- ConsumerGroupId: int(createRandomUInt32()),
- Name: createRandomString(16),
- }
- err := client.CreateConsumerGroup(request)
+ groupId := createRandomUInt32()
+ _, err := client.CreateConsumerGroup(
+ iggcon.NewIdentifier(streamId),
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ createRandomString(16),
+ &groupId,
+ )
itShouldReturnSpecificError(err, "topic_id_not_found")
})
@@ -78,13 +77,13 @@ var _ = Describe("CREATE CONSUMER GROUP:", func() {
topicId, _ := successfullyCreateTopic(streamId, client)
_, name := successfullyCreateConsumer(streamId, topicId, client)
- request := iggcon.CreateConsumerGroupRequest{
- StreamId: iggcon.NewIdentifier(streamId),
- TopicId: iggcon.NewIdentifier(topicId),
- ConsumerGroupId: int(createRandomUInt32()),
- Name: name,
- }
- err := client.CreateConsumerGroup(request)
+ groupId := createRandomUInt32()
+ _, err := client.CreateConsumerGroup(
+ iggcon.NewIdentifier(streamId),
+ iggcon.NewIdentifier(topicId),
+ name,
+ &groupId,
+ )
itShouldReturnSpecificError(err, "cannot_create_consumer_groups_directory")
})
@@ -96,13 +95,12 @@ var _ = Describe("CREATE CONSUMER GROUP:", func() {
topicId, _ := successfullyCreateTopic(streamId, client)
groupId, _ := successfullyCreateConsumer(streamId, topicId, client)
- request := iggcon.CreateConsumerGroupRequest{
- StreamId: iggcon.NewIdentifier(streamId),
- TopicId: iggcon.NewIdentifier(topicId),
- ConsumerGroupId: groupId,
- Name: createRandomString(16),
- }
- err := client.CreateConsumerGroup(request)
+ uint32GroupId := uint32(groupId)
+ _, err := client.CreateConsumerGroup(
+ iggcon.NewIdentifier(streamId),
+ iggcon.NewIdentifier(topicId),
+ createRandomString(16),
+ &uint32GroupId)
itShouldReturnSpecificError(err, "consumer_group_already_exists")
})
@@ -113,13 +111,11 @@ var _ = Describe("CREATE CONSUMER GROUP:", func() {
defer deleteStreamAfterTests(streamId, client)
topicId, _ := successfullyCreateTopic(streamId, client)
- request := iggcon.CreateConsumerGroupRequest{
- StreamId: iggcon.NewIdentifier(streamId),
- TopicId: iggcon.NewIdentifier(topicId),
- ConsumerGroupId: int(createRandomUInt32()),
- Name: createRandomString(256),
- }
- err := client.CreateConsumerGroup(request)
+ groupId := createRandomUInt32()
+ _, err := client.CreateConsumerGroup(iggcon.NewIdentifier(streamId),
+ iggcon.NewIdentifier(topicId),
+ createRandomString(256),
+ &groupId)
itShouldReturnSpecificError(err, "consumer_group_name_too_long")
})
@@ -127,14 +123,14 @@ var _ = Describe("CREATE CONSUMER GROUP:", func() {
When("User is not logged in", func() {
Context("and tries to create consumer group", func() {
- client := createConnection()
- request := iggcon.CreateConsumerGroupRequest{
- StreamId: iggcon.NewIdentifier(int(createRandomUInt32())),
- TopicId: iggcon.NewIdentifier(int(createRandomUInt32())),
- ConsumerGroupId: int(createRandomUInt32()),
- Name: createRandomString(16),
- }
- err := client.CreateConsumerGroup(request)
+ client := createClient()
+ groupId := createRandomUInt32()
+ _, err := client.CreateConsumerGroup(
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ createRandomString(16),
+ &groupId,
+ )
itShouldReturnUnauthenticatedError(err)
})
diff --git a/bdd/go/tests/tcp_test/consumers_feature_delete.go b/bdd/go/tests/tcp_test/consumers_feature_delete.go
index a298c38a..d28a384b 100644
--- a/bdd/go/tests/tcp_test/consumers_feature_delete.go
+++ b/bdd/go/tests/tcp_test/consumers_feature_delete.go
@@ -32,11 +32,11 @@ var _ = Describe("DELETE CONSUMER GROUP:", func() {
defer deleteStreamAfterTests(streamId, client)
topicId, _ := successfullyCreateTopic(streamId, client)
groupId, _ := successfullyCreateConsumer(streamId, topicId, client)
- err := client.DeleteConsumerGroup(iggcon.DeleteConsumerGroupRequest{
- StreamId: iggcon.NewIdentifier(streamId),
- TopicId: iggcon.NewIdentifier(topicId),
- ConsumerGroupId: iggcon.NewIdentifier(groupId),
- })
+ err := client.DeleteConsumerGroup(
+ iggcon.NewIdentifier(streamId),
+ iggcon.NewIdentifier(topicId),
+ iggcon.NewIdentifier(groupId),
+ )
itShouldNotReturnError(err)
itShouldSuccessfullyDeletedConsumer(streamId, topicId, groupId, client)
@@ -48,11 +48,11 @@ var _ = Describe("DELETE CONSUMER GROUP:", func() {
defer deleteStreamAfterTests(streamId, client)
topicId, _ := successfullyCreateTopic(streamId, client)
groupId := int(createRandomUInt32())
- err := client.DeleteConsumerGroup(iggcon.DeleteConsumerGroupRequest{
- StreamId: iggcon.NewIdentifier(streamId),
- TopicId: iggcon.NewIdentifier(topicId),
- ConsumerGroupId: iggcon.NewIdentifier(groupId),
- })
+ err := client.DeleteConsumerGroup(
+ iggcon.NewIdentifier(streamId),
+ iggcon.NewIdentifier(topicId),
+ iggcon.NewIdentifier(groupId),
+ )
itShouldReturnSpecificIggyError(err, ierror.ConsumerGroupIdNotFound)
})
@@ -63,11 +63,11 @@ var _ = Describe("DELETE CONSUMER GROUP:", func() {
defer deleteStreamAfterTests(streamId, client)
topicId := int(createRandomUInt32())
- err := client.DeleteConsumerGroup(iggcon.DeleteConsumerGroupRequest{
- StreamId: iggcon.NewIdentifier(streamId),
- TopicId: iggcon.NewIdentifier(topicId),
- ConsumerGroupId: iggcon.NewIdentifier(int(createRandomUInt32())),
- })
+ err := client.DeleteConsumerGroup(
+ iggcon.NewIdentifier(streamId),
+ iggcon.NewIdentifier(topicId),
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ )
itShouldReturnSpecificError(err, "topic_id_not_found")
})
@@ -77,11 +77,11 @@ var _ = Describe("DELETE CONSUMER GROUP:", func() {
streamId := int(createRandomUInt32())
topicId := int(createRandomUInt32())
- err := client.DeleteConsumerGroup(iggcon.DeleteConsumerGroupRequest{
- StreamId: iggcon.NewIdentifier(streamId),
- TopicId: iggcon.NewIdentifier(topicId),
- ConsumerGroupId: iggcon.NewIdentifier(int(createRandomUInt32())),
- })
+ err := client.DeleteConsumerGroup(
+ iggcon.NewIdentifier(streamId),
+ iggcon.NewIdentifier(topicId),
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ )
itShouldReturnSpecificError(err, "stream_id_not_found")
})
@@ -89,12 +89,12 @@ var _ = Describe("DELETE CONSUMER GROUP:", func() {
When("User is not logged in", func() {
Context("and tries to delete consumer group", func() {
- client := createConnection()
- err := client.DeleteConsumerGroup(iggcon.DeleteConsumerGroupRequest{
- StreamId: iggcon.NewIdentifier(int(createRandomUInt32())),
- TopicId: iggcon.NewIdentifier(int(createRandomUInt32())),
- ConsumerGroupId: iggcon.NewIdentifier(int(createRandomUInt32())),
- })
+ client := createClient()
+ err := client.DeleteConsumerGroup(
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ )
itShouldReturnUnauthenticatedError(err)
})
diff --git a/bdd/go/tests/tcp_test/consumers_feature_get_all.go b/bdd/go/tests/tcp_test/consumers_feature_get_all.go
index c4f0f921..29508311 100644
--- a/bdd/go/tests/tcp_test/consumers_feature_get_all.go
+++ b/bdd/go/tests/tcp_test/consumers_feature_get_all.go
@@ -40,7 +40,7 @@ var _ = Describe("GET ALL CONSUMER GROUPS:", func() {
When("User is not logged in", func() {
Context("and tries to get all consumer groups", func() {
- client := createConnection()
+ client := createClient()
_, err := client.GetConsumerGroups(iggcon.NewIdentifier(int(createRandomUInt32())), iggcon.NewIdentifier(int(createRandomUInt32())))
itShouldReturnUnauthenticatedError(err)
diff --git a/bdd/go/tests/tcp_test/consumers_feature_get_by_id.go b/bdd/go/tests/tcp_test/consumers_feature_get_by_id.go
index 95072fb1..b033edd2 100644
--- a/bdd/go/tests/tcp_test/consumers_feature_get_by_id.go
+++ b/bdd/go/tests/tcp_test/consumers_feature_get_by_id.go
@@ -24,7 +24,7 @@ import (
)
var _ = Describe("GET CONSUMER GROUP BY ID:", func() {
- prefix := "GetConsumerGroupById"
+ prefix := "GetConsumerGroup"
When("User is logged in", func() {
Context("and tries to get existing consumer group", func() {
client := createAuthorizedConnection()
@@ -32,16 +32,16 @@ var _ = Describe("GET CONSUMER GROUP BY ID:", func() {
defer deleteStreamAfterTests(streamId, client)
topicId, _ := successfullyCreateTopic(streamId, client)
groupId, name := successfullyCreateConsumer(streamId, topicId, client)
- group, err := client.GetConsumerGroupById(iggcon.NewIdentifier(streamId), iggcon.NewIdentifier(topicId), iggcon.NewIdentifier(groupId))
+ group, err := client.GetConsumerGroup(iggcon.NewIdentifier(streamId), iggcon.NewIdentifier(topicId), iggcon.NewIdentifier(groupId))
itShouldNotReturnError(err)
- itShouldReturnSpecificConsumer(groupId, name, group)
+ itShouldReturnSpecificConsumer(groupId, name, &group.ConsumerGroup)
})
Context("and tries to get consumer from non-existing stream", func() {
client := createAuthorizedConnection()
- _, err := client.GetConsumerGroupById(
+ _, err := client.GetConsumerGroup(
iggcon.NewIdentifier(int(createRandomUInt32())),
iggcon.NewIdentifier(int(createRandomUInt32())),
iggcon.NewIdentifier(int(createRandomUInt32())))
@@ -54,7 +54,7 @@ var _ = Describe("GET CONSUMER GROUP BY ID:", func() {
streamId, _ := successfullyCreateStream(prefix, client)
defer deleteStreamAfterTests(streamId, client)
- _, err := client.GetConsumerGroupById(
+ _, err := client.GetConsumerGroup(
iggcon.NewIdentifier(streamId),
iggcon.NewIdentifier(int(createRandomUInt32())),
iggcon.NewIdentifier(int(createRandomUInt32())))
@@ -68,7 +68,7 @@ var _ = Describe("GET CONSUMER GROUP BY ID:", func() {
defer deleteStreamAfterTests(streamId, client)
topicId, _ := successfullyCreateTopic(streamId, client)
- _, err := client.GetConsumerGroupById(
+ _, err := client.GetConsumerGroup(
iggcon.NewIdentifier(streamId),
iggcon.NewIdentifier(topicId),
iggcon.NewIdentifier(int(createRandomUInt32())))
diff --git a/bdd/go/tests/tcp_test/consumers_feature_join.go b/bdd/go/tests/tcp_test/consumers_feature_join.go
index 3a992f2b..cebf9101 100644
--- a/bdd/go/tests/tcp_test/consumers_feature_join.go
+++ b/bdd/go/tests/tcp_test/consumers_feature_join.go
@@ -31,11 +31,11 @@ var _ = Describe("JOIN CONSUMER GROUP:", func() {
defer deleteStreamAfterTests(streamId, client)
topicId, _ := successfullyCreateTopic(streamId, client)
groupId, _ := successfullyCreateConsumer(streamId, topicId, client)
- err := client.JoinConsumerGroup(iggcon.JoinConsumerGroupRequest{
- StreamId: iggcon.NewIdentifier(streamId),
- TopicId: iggcon.NewIdentifier(topicId),
- ConsumerGroupId: iggcon.NewIdentifier(groupId),
- })
+ err := client.JoinConsumerGroup(
+ iggcon.NewIdentifier(streamId),
+ iggcon.NewIdentifier(topicId),
+ iggcon.NewIdentifier(groupId),
+ )
itShouldNotReturnError(err)
itShouldSuccessfullyJoinConsumer(streamId, topicId, groupId, client)
@@ -47,11 +47,10 @@ var _ = Describe("JOIN CONSUMER GROUP:", func() {
defer deleteStreamAfterTests(streamId, client)
topicId, _ := successfullyCreateTopic(streamId, client)
groupId := int(createRandomUInt32())
- err := client.JoinConsumerGroup(iggcon.JoinConsumerGroupRequest{
- StreamId: iggcon.NewIdentifier(streamId),
- TopicId: iggcon.NewIdentifier(topicId),
- ConsumerGroupId: iggcon.NewIdentifier(groupId),
- })
+ err := client.JoinConsumerGroup(iggcon.NewIdentifier(streamId),
+ iggcon.NewIdentifier(topicId),
+ iggcon.NewIdentifier(groupId),
+ )
itShouldReturnSpecificError(err, "consumer_group_not_found")
})
@@ -62,11 +61,11 @@ var _ = Describe("JOIN CONSUMER GROUP:", func() {
defer deleteStreamAfterTests(streamId, client)
topicId := int(createRandomUInt32())
- err := client.JoinConsumerGroup(iggcon.JoinConsumerGroupRequest{
- StreamId: iggcon.NewIdentifier(streamId),
- TopicId: iggcon.NewIdentifier(topicId),
- ConsumerGroupId: iggcon.NewIdentifier(int(createRandomUInt32())),
- })
+ err := client.JoinConsumerGroup(
+ iggcon.NewIdentifier(streamId),
+ iggcon.NewIdentifier(topicId),
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ )
itShouldReturnSpecificError(err, "topic_id_not_found")
})
@@ -76,11 +75,11 @@ var _ = Describe("JOIN CONSUMER GROUP:", func() {
streamId := int(createRandomUInt32())
topicId := int(createRandomUInt32())
- err := client.JoinConsumerGroup(iggcon.JoinConsumerGroupRequest{
- StreamId: iggcon.NewIdentifier(streamId),
- TopicId: iggcon.NewIdentifier(topicId),
- ConsumerGroupId: iggcon.NewIdentifier(int(createRandomUInt32())),
- })
+ err := client.JoinConsumerGroup(
+ iggcon.NewIdentifier(streamId),
+ iggcon.NewIdentifier(topicId),
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ )
itShouldReturnSpecificError(err, "stream_id_not_found")
})
@@ -88,12 +87,12 @@ var _ = Describe("JOIN CONSUMER GROUP:", func() {
When("User is not logged in", func() {
Context("and tries to join to the consumer group", func() {
- client := createConnection()
- err := client.JoinConsumerGroup(iggcon.JoinConsumerGroupRequest{
- StreamId: iggcon.NewIdentifier(int(createRandomUInt32())),
- TopicId: iggcon.NewIdentifier(int(createRandomUInt32())),
- ConsumerGroupId: iggcon.NewIdentifier(int(createRandomUInt32())),
- })
+ client := createClient()
+ err := client.JoinConsumerGroup(
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ )
itShouldReturnUnauthenticatedError(err)
})
diff --git a/bdd/go/tests/tcp_test/consumers_feature_leave.go b/bdd/go/tests/tcp_test/consumers_feature_leave.go
index 4ba7fc76..f3ff081b 100644
--- a/bdd/go/tests/tcp_test/consumers_feature_leave.go
+++ b/bdd/go/tests/tcp_test/consumers_feature_leave.go
@@ -33,11 +33,11 @@ var _ = Describe("LEAVE CONSUMER GROUP:", func() {
groupId, _ := successfullyCreateConsumer(streamId, topicId, client)
successfullyJoinConsumer(streamId, topicId, groupId, client)
- err := client.LeaveConsumerGroup(iggcon.LeaveConsumerGroupRequest{
- StreamId: iggcon.NewIdentifier(streamId),
- TopicId: iggcon.NewIdentifier(topicId),
- ConsumerGroupId: iggcon.NewIdentifier(groupId),
- })
+ err := client.LeaveConsumerGroup(
+ iggcon.NewIdentifier(streamId),
+ iggcon.NewIdentifier(topicId),
+ iggcon.NewIdentifier(groupId),
+ )
itShouldNotReturnError(err)
itShouldSuccessfullyLeaveConsumer(streamId, topicId, groupId, client)
@@ -49,11 +49,11 @@ var _ = Describe("LEAVE CONSUMER GROUP:", func() {
defer deleteStreamAfterTests(streamId, client)
topicId, _ := successfullyCreateTopic(streamId, client)
groupId := int(createRandomUInt32())
- err := client.LeaveConsumerGroup(iggcon.LeaveConsumerGroupRequest{
- StreamId: iggcon.NewIdentifier(streamId),
- TopicId: iggcon.NewIdentifier(topicId),
- ConsumerGroupId: iggcon.NewIdentifier(groupId),
- })
+ err := client.LeaveConsumerGroup(
+ iggcon.NewIdentifier(streamId),
+ iggcon.NewIdentifier(topicId),
+ iggcon.NewIdentifier(groupId),
+ )
itShouldReturnSpecificError(err, "consumer_group_not_found")
})
@@ -64,11 +64,11 @@ var _ = Describe("LEAVE CONSUMER GROUP:", func() {
defer deleteStreamAfterTests(streamId, client)
topicId := int(createRandomUInt32())
- err := client.LeaveConsumerGroup(iggcon.LeaveConsumerGroupRequest{
- StreamId: iggcon.NewIdentifier(streamId),
- TopicId: iggcon.NewIdentifier(topicId),
- ConsumerGroupId: iggcon.NewIdentifier(int(createRandomUInt32())),
- })
+ err := client.LeaveConsumerGroup(
+ iggcon.NewIdentifier(streamId),
+ iggcon.NewIdentifier(topicId),
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ )
itShouldReturnSpecificError(err, "topic_id_not_found")
})
@@ -78,11 +78,11 @@ var _ = Describe("LEAVE CONSUMER GROUP:", func() {
streamId := int(createRandomUInt32())
topicId := int(createRandomUInt32())
- err := client.LeaveConsumerGroup(iggcon.LeaveConsumerGroupRequest{
- StreamId: iggcon.NewIdentifier(streamId),
- TopicId: iggcon.NewIdentifier(topicId),
- ConsumerGroupId: iggcon.NewIdentifier(int(createRandomUInt32())),
- })
+ err := client.LeaveConsumerGroup(
+ iggcon.NewIdentifier(streamId),
+ iggcon.NewIdentifier(topicId),
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ )
itShouldReturnSpecificError(err, "stream_id_not_found")
})
@@ -90,12 +90,12 @@ var _ = Describe("LEAVE CONSUMER GROUP:", func() {
When("User is not logged in", func() {
Context("and tries to leave to the consumer group", func() {
- client := createConnection()
- err := client.LeaveConsumerGroup(iggcon.LeaveConsumerGroupRequest{
- StreamId: iggcon.NewIdentifier(int(createRandomUInt32())),
- TopicId: iggcon.NewIdentifier(int(createRandomUInt32())),
- ConsumerGroupId: iggcon.NewIdentifier(int(createRandomUInt32())),
- })
+ client := createClient()
+ err := client.LeaveConsumerGroup(
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ )
itShouldReturnUnauthenticatedError(err)
})
diff --git a/bdd/go/tests/tcp_test/consumers_steps.go b/bdd/go/tests/tcp_test/consumers_steps.go
index 6dc2eb19..5ea8b1fa 100644
--- a/bdd/go/tests/tcp_test/consumers_steps.go
+++ b/bdd/go/tests/tcp_test/consumers_steps.go
@@ -20,34 +20,34 @@ package tcp_test
import (
"strconv"
- "github.com/apache/iggy/foreign/go"
iggcon "github.com/apache/iggy/foreign/go/contracts"
+ "github.com/apache/iggy/foreign/go/iggycli"
. "github.com/onsi/ginkgo/v2"
. "github.com/onsi/gomega"
)
// operations
-func successfullyCreateConsumer(streamId int, topicId int, client iggy.MessageStream) (int, string) {
- request := iggcon.CreateConsumerGroupRequest{
- StreamId: iggcon.NewIdentifier(streamId),
- TopicId: iggcon.NewIdentifier(topicId),
- ConsumerGroupId: int(createRandomUInt32()),
- Name: createRandomString(16),
- }
- err := client.CreateConsumerGroup(request)
-
- itShouldSuccessfullyCreateConsumer(streamId, topicId, request.ConsumerGroupId, request.Name, client)
+func successfullyCreateConsumer(streamId int, topicId int, cli iggycli.Client) (int, string) {
+ groupId := createRandomUInt32()
+ name := createRandomString(16)
+ _, err := cli.CreateConsumerGroup(iggcon.NewIdentifier(streamId),
+ iggcon.NewIdentifier(topicId),
+ name,
+ &groupId,
+ )
+
+ itShouldSuccessfullyCreateConsumer(streamId, topicId, int(groupId), name, cli)
itShouldNotReturnError(err)
- return request.ConsumerGroupId, request.Name
+ return int(groupId), name
}
-func successfullyJoinConsumer(streamId int, topicId int, groupId int, client iggy.MessageStream) {
- request := iggcon.JoinConsumerGroupRequest{
- StreamId: iggcon.NewIdentifier(streamId),
- TopicId: iggcon.NewIdentifier(topicId),
- ConsumerGroupId: iggcon.NewIdentifier(groupId),
- }
- err := client.JoinConsumerGroup(request)
+func successfullyJoinConsumer(streamId int, topicId int, groupId int, client iggycli.Client) {
+
+ err := client.JoinConsumerGroup(
+ iggcon.NewIdentifier(streamId),
+ iggcon.NewIdentifier(topicId),
+ iggcon.NewIdentifier(groupId),
+ )
itShouldSuccessfullyJoinConsumer(streamId, topicId, groupId, client)
itShouldNotReturnError(err)
@@ -55,7 +55,7 @@ func successfullyJoinConsumer(streamId int, topicId int, groupId int, client igg
//assertions
-func itShouldReturnSpecificConsumer(id int, name string, consumer *iggcon.ConsumerGroupResponse) {
+func itShouldReturnSpecificConsumer(id int, name string, consumer *iggcon.ConsumerGroup) {
It("should fetch consumer with id "+string(rune(id)), func() {
Expect(consumer).NotTo(BeNil())
Expect(consumer.Id).To(Equal(id))
@@ -67,12 +67,12 @@ func itShouldReturnSpecificConsumer(id int, name string, consumer *iggcon.Consum
})
}
-func itShouldContainSpecificConsumer(id int, name string, consumers []iggcon.ConsumerGroupResponse) {
+func itShouldContainSpecificConsumer(id int, name string, consumers []iggcon.ConsumerGroup) {
It("should fetch at least one consumer", func() {
Expect(len(consumers)).NotTo(Equal(0))
})
- var consumer iggcon.ConsumerGroupResponse
+ var consumer iggcon.ConsumerGroup
found := false
for _, s := range consumers {
@@ -94,8 +94,8 @@ func itShouldContainSpecificConsumer(id int, name string, consumers []iggcon.Con
})
}
-func itShouldSuccessfullyCreateConsumer(streamId int, topicId int, groupId int, expectedName string, client iggy.MessageStream) {
- consumer, err := client.GetConsumerGroupById(iggcon.NewIdentifier(streamId), iggcon.NewIdentifier(topicId), iggcon.NewIdentifier(groupId))
+func itShouldSuccessfullyCreateConsumer(streamId int, topicId int, groupId int, expectedName string, client iggycli.Client) {
+ consumer, err := client.GetConsumerGroup(iggcon.NewIdentifier(streamId), iggcon.NewIdentifier(topicId), iggcon.NewIdentifier(groupId))
It("should create consumer with id "+string(rune(groupId)), func() {
Expect(consumer).NotTo(BeNil())
@@ -109,8 +109,8 @@ func itShouldSuccessfullyCreateConsumer(streamId int, topicId int, groupId int,
itShouldNotReturnError(err)
}
-func itShouldSuccessfullyDeletedConsumer(streamId int, topicId int, groupId int, client iggy.MessageStream) {
- consumer, err := client.GetConsumerGroupById(iggcon.NewIdentifier(streamId), iggcon.NewIdentifier(topicId), iggcon.NewIdentifier(groupId))
+func itShouldSuccessfullyDeletedConsumer(streamId int, topicId int, groupId int, client iggycli.Client) {
+ consumer, err := client.GetConsumerGroup(iggcon.NewIdentifier(streamId), iggcon.NewIdentifier(topicId), iggcon.NewIdentifier(groupId))
itShouldReturnSpecificError(err, "consumer_group_not_found")
It("should not return consumer", func() {
@@ -118,8 +118,8 @@ func itShouldSuccessfullyDeletedConsumer(streamId int, topicId int, groupId int,
})
}
-func itShouldSuccessfullyJoinConsumer(streamId int, topicId int, groupId int, client iggy.MessageStream) {
- consumer, err := client.GetConsumerGroupById(iggcon.NewIdentifier(streamId), iggcon.NewIdentifier(topicId), iggcon.NewIdentifier(groupId))
+func itShouldSuccessfullyJoinConsumer(streamId int, topicId int, groupId int, client iggycli.Client) {
+ consumer, err := client.GetConsumerGroup(iggcon.NewIdentifier(streamId), iggcon.NewIdentifier(topicId), iggcon.NewIdentifier(groupId))
It("should join consumer with id "+string(rune(groupId)), func() {
Expect(consumer).NotTo(BeNil())
@@ -129,8 +129,8 @@ func itShouldSuccessfullyJoinConsumer(streamId int, topicId int, groupId int, cl
itShouldNotReturnError(err)
}
-func itShouldSuccessfullyLeaveConsumer(streamId int, topicId int, groupId int, client iggy.MessageStream) {
- consumer, err := client.GetConsumerGroupById(iggcon.NewIdentifier(streamId), iggcon.NewIdentifier(topicId), iggcon.NewIdentifier(groupId))
+func itShouldSuccessfullyLeaveConsumer(streamId int, topicId int, groupId int, client iggycli.Client) {
+ consumer, err := client.GetConsumerGroup(iggcon.NewIdentifier(streamId), iggcon.NewIdentifier(topicId), iggcon.NewIdentifier(groupId))
It("should leave consumer with id "+string(rune(groupId)), func() {
Expect(consumer).NotTo(BeNil())
diff --git a/bdd/go/tests/tcp_test/messages_feature_send.go b/bdd/go/tests/tcp_test/messages_feature_send.go
index ef8f549b..746e1168 100644
--- a/bdd/go/tests/tcp_test/messages_feature_send.go
+++ b/bdd/go/tests/tcp_test/messages_feature_send.go
@@ -31,13 +31,12 @@ var _ = Describe("SEND MESSAGES:", func() {
defer deleteStreamAfterTests(streamId, client)
topicId, _ := successfullyCreateTopic(streamId, client)
messages := createDefaultMessages()
- request := iggcon.SendMessagesRequest{
- StreamId: iggcon.NewIdentifier(streamId),
- TopicId: iggcon.NewIdentifier(topicId),
- Partitioning: iggcon.None(),
- Messages: messages,
- }
- err := client.SendMessages(request)
+ err := client.SendMessages(
+ iggcon.NewIdentifier(streamId),
+ iggcon.NewIdentifier(topicId),
+ iggcon.None(),
+ messages,
+ )
itShouldNotReturnError(err)
itShouldSuccessfullyPublishMessages(streamId, topicId, messages, client)
})
@@ -47,26 +46,24 @@ var _ = Describe("SEND MESSAGES:", func() {
streamId, _ := successfullyCreateStream("2"+prefix, client)
defer deleteStreamAfterTests(streamId, client)
messages := createDefaultMessages()
- request := iggcon.SendMessagesRequest{
- StreamId: iggcon.NewIdentifier(streamId),
- TopicId: iggcon.NewIdentifier(int(createRandomUInt32())),
- Partitioning: iggcon.None(),
- Messages: messages,
- }
- err := client.SendMessages(request)
+ err := client.SendMessages(
+ iggcon.NewIdentifier(streamId),
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ iggcon.None(),
+ messages,
+ )
itShouldReturnSpecificError(err, "topic_id_not_found")
})
Context("and tries to send messages to the non existing stream", func() {
client := createAuthorizedConnection()
messages := createDefaultMessages()
- request := iggcon.SendMessagesRequest{
- StreamId: iggcon.NewIdentifier(int(createRandomUInt32())),
- TopicId: iggcon.NewIdentifier(int(createRandomUInt32())),
- Partitioning: iggcon.None(),
- Messages: messages,
- }
- err := client.SendMessages(request)
+ err := client.SendMessages(
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ iggcon.None(),
+ messages,
+ )
itShouldReturnSpecificError(err, "stream_id_not_found")
})
@@ -76,13 +73,12 @@ var _ = Describe("SEND MESSAGES:", func() {
defer deleteStreamAfterTests(streamId, client)
topicId, _ := successfullyCreateTopic(streamId, client)
messages := createDefaultMessages()
- request := iggcon.SendMessagesRequest{
- StreamId: iggcon.NewIdentifier(streamId),
- TopicId: iggcon.NewIdentifier(topicId),
- Partitioning: iggcon.PartitionId(int(createRandomUInt32())),
- Messages: messages,
- }
- err := client.SendMessages(request)
+ err := client.SendMessages(
+ iggcon.NewIdentifier(streamId),
+ iggcon.NewIdentifier(topicId),
+ iggcon.PartitionId(int(createRandomUInt32())),
+ messages,
+ )
itShouldReturnSpecificError(err, "partition_not_found")
})
@@ -91,28 +87,26 @@ var _ = Describe("SEND MESSAGES:", func() {
streamId, _ := successfullyCreateStream(prefix, client)
defer deleteStreamAfterTests(streamId, createAuthorizedConnection())
topicId, _ := successfullyCreateTopic(streamId, client)
- request := iggcon.SendMessagesRequest{
- StreamId: iggcon.NewIdentifier(streamId),
- TopicId: iggcon.NewIdentifier(topicId),
- Partitioning: iggcon.PartitionId(int(createRandomUInt32())),
- Messages: []iggcon.IggyMessage{},
- }
- err := client.SendMessages(request)
+ err := client.SendMessages(
+ iggcon.NewIdentifier(streamId),
+ iggcon.NewIdentifier(topicId),
+ iggcon.PartitionId(int(createRandomUInt32())),
+ []iggcon.IggyMessage{},
+ )
itShouldReturnSpecificError(err, "messages_count_should_be_greater_than_zero")
})
})
When("User is not logged in", func() {
Context("and tries to update stream", func() {
- client := createConnection()
+ client := createClient()
messages := createDefaultMessages()
- request := iggcon.SendMessagesRequest{
- StreamId: iggcon.NewIdentifier(int(createRandomUInt32())),
- TopicId: iggcon.NewIdentifier(int(createRandomUInt32())),
- Partitioning: iggcon.None(),
- Messages: messages,
- }
- err := client.SendMessages(request)
+ err := client.SendMessages(
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ iggcon.None(),
+ messages,
+ )
itShouldReturnUnauthenticatedError(err)
})
diff --git a/bdd/go/tests/tcp_test/messages_steps.go b/bdd/go/tests/tcp_test/messages_steps.go
index 3f733a49..ed30acb8 100644
--- a/bdd/go/tests/tcp_test/messages_steps.go
+++ b/bdd/go/tests/tcp_test/messages_steps.go
@@ -21,8 +21,8 @@ import (
"bytes"
"reflect"
- "github.com/apache/iggy/foreign/go"
iggcon "github.com/apache/iggy/foreign/go/contracts"
+ "github.com/apache/iggy/foreign/go/iggycli"
"github.com/google/uuid"
. "github.com/onsi/ginkgo/v2"
. "github.com/onsi/gomega"
@@ -42,18 +42,18 @@ func createDefaultMessages() []iggcon.IggyMessage {
return []iggcon.IggyMessage{msg1, msg2}
}
-func itShouldSuccessfullyPublishMessages(streamId int, topicId int, messages []iggcon.IggyMessage, client iggy.MessageStream) {
- result, err := client.PollMessages(iggcon.FetchMessagesRequest{
- StreamId: iggcon.NewIdentifier(streamId),
- TopicId: iggcon.NewIdentifier(topicId),
- Consumer: iggcon.Consumer{
- Kind: iggcon.ConsumerSingle,
+func itShouldSuccessfullyPublishMessages(streamId int, topicId int, messages []iggcon.IggyMessage, client iggycli.Client) {
+ result, err := client.PollMessages(
+ iggcon.NewIdentifier(streamId),
+ iggcon.NewIdentifier(topicId),
+ iggcon.Consumer{
+ Kind: iggcon.ConsumerKindSingle,
Id: iggcon.NewIdentifier(int(createRandomUInt32())),
},
- PollingStrategy: iggcon.FirstPollingStrategy(),
- Count: len(messages),
- AutoCommit: true,
- })
+ iggcon.FirstPollingStrategy(),
+ uint32(len(messages)),
+ true,
+ nil)
It("It should not be nil", func() {
Expect(result).NotTo(BeNil())
diff --git a/bdd/go/tests/tcp_test/partitions_feature_create.go b/bdd/go/tests/tcp_test/partitions_feature_create.go
index 64327ce6..c981ec75 100644
--- a/bdd/go/tests/tcp_test/partitions_feature_create.go
+++ b/bdd/go/tests/tcp_test/partitions_feature_create.go
@@ -23,33 +23,30 @@ import (
)
var _ = Describe("CREATE PARTITION:", func() {
- prefix := "CreatePartition"
+ prefix := "CreatePartitions"
When("User is logged in", func() {
Context("and tries to create partitions for existing stream", func() {
client := createAuthorizedConnection()
streamId, _ := successfullyCreateStream(prefix, client)
defer deleteStreamAfterTests(streamId, client)
topicId, _ := successfullyCreateTopic(streamId, client)
-
- request := iggcon.CreatePartitionsRequest{
- StreamId: iggcon.NewIdentifier(streamId),
- TopicId: iggcon.NewIdentifier(topicId),
- PartitionsCount: 10,
- }
- err := client.CreatePartition(request)
+ partitionsCount := 10
+ err := client.CreatePartitions(
+ iggcon.NewIdentifier(streamId),
+ iggcon.NewIdentifier(topicId),
+ uint32(partitionsCount))
itShouldNotReturnError(err)
- itShouldHaveExpectedNumberOfPartitions(streamId, topicId, request.PartitionsCount+2, client)
+ itShouldHaveExpectedNumberOfPartitions(streamId, topicId, partitionsCount+2, client)
})
Context("and tries to create partitions for a non existing stream", func() {
client := createAuthorizedConnection()
- request := iggcon.CreatePartitionsRequest{
- StreamId: iggcon.NewIdentifier(int(createRandomUInt32())),
- TopicId: iggcon.NewIdentifier(int(createRandomUInt32())),
- PartitionsCount: 10,
- }
- err := client.CreatePartition(request)
+ err := client.CreatePartitions(
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ 10,
+ )
itShouldReturnSpecificError(err, "stream_id_not_found")
})
@@ -58,12 +55,11 @@ var _ = Describe("CREATE PARTITION:", func() {
client := createAuthorizedConnection()
streamId, _ := successfullyCreateStream(prefix, client)
defer deleteStreamAfterTests(streamId, client)
- request := iggcon.CreatePartitionsRequest{
- StreamId: iggcon.NewIdentifier(streamId),
- TopicId: iggcon.NewIdentifier(int(createRandomUInt32())),
- PartitionsCount: 10,
- }
- err := client.CreatePartition(request)
+ err := client.CreatePartitions(
+ iggcon.NewIdentifier(streamId),
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ 10,
+ )
itShouldReturnSpecificError(err, "topic_id_not_found")
})
@@ -71,13 +67,12 @@ var _ = Describe("CREATE PARTITION:", func() {
When("User is not logged in", func() {
Context("and tries to create partitions", func() {
- client := createConnection()
- request := iggcon.CreatePartitionsRequest{
- StreamId: iggcon.NewIdentifier(int(createRandomUInt32())),
- TopicId: iggcon.NewIdentifier(int(createRandomUInt32())),
- PartitionsCount: 10,
- }
- err := client.CreatePartition(request)
+ client := createClient()
+ err := client.CreatePartitions(
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ 10,
+ )
itShouldReturnUnauthenticatedError(err)
})
diff --git a/bdd/go/tests/tcp_test/partitions_feature_delete.go b/bdd/go/tests/tcp_test/partitions_feature_delete.go
index 4d011138..3a571db6 100644
--- a/bdd/go/tests/tcp_test/partitions_feature_delete.go
+++ b/bdd/go/tests/tcp_test/partitions_feature_delete.go
@@ -23,7 +23,7 @@ import (
)
var _ = Describe("DELETE PARTITION:", func() {
- prefix := "DeletePartition"
+ prefix := "DeletePartitions"
When("User is logged in", func() {
Context("and tries to delete partitions for existing stream", func() {
client := createAuthorizedConnection()
@@ -31,25 +31,24 @@ var _ = Describe("DELETE PARTITION:", func() {
defer deleteStreamAfterTests(streamId, client)
topicId, _ := successfullyCreateTopic(streamId, client)
- request := iggcon.DeletePartitionRequest{
- StreamId: iggcon.NewIdentifier(streamId),
- TopicId: iggcon.NewIdentifier(topicId),
- PartitionsCount: 1,
- }
- err := client.DeletePartition(request)
+ partitionsCount := 1
+ err := client.DeletePartitions(
+ iggcon.NewIdentifier(streamId),
+ iggcon.NewIdentifier(topicId),
+ 1,
+ )
itShouldNotReturnError(err)
- itShouldHaveExpectedNumberOfPartitions(streamId, topicId, 2-request.PartitionsCount, client)
+ itShouldHaveExpectedNumberOfPartitions(streamId, topicId, 2-partitionsCount, client)
})
Context("and tries to delete partitions for a non existing stream", func() {
client := createAuthorizedConnection()
- request := iggcon.DeletePartitionRequest{
- StreamId: iggcon.NewIdentifier(int(createRandomUInt32())),
- TopicId: iggcon.NewIdentifier(int(createRandomUInt32())),
- PartitionsCount: 10,
- }
- err := client.DeletePartition(request)
+ err := client.DeletePartitions(
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ 10,
+ )
itShouldReturnSpecificError(err, "stream_id_not_found")
})
@@ -58,12 +57,11 @@ var _ = Describe("DELETE PARTITION:", func() {
client := createAuthorizedConnection()
streamId, _ := successfullyCreateStream(prefix, client)
defer deleteStreamAfterTests(streamId, client)
- request := iggcon.DeletePartitionRequest{
- StreamId: iggcon.NewIdentifier(streamId),
- TopicId: iggcon.NewIdentifier(int(createRandomUInt32())),
- PartitionsCount: 10,
- }
- err := client.DeletePartition(request)
+ err := client.DeletePartitions(
+ iggcon.NewIdentifier(streamId),
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ 10,
+ )
itShouldReturnSpecificError(err, "topic_id_not_found")
})
@@ -71,13 +69,12 @@ var _ = Describe("DELETE PARTITION:", func() {
When("User is not logged in", func() {
Context("and tries to delete partitions", func() {
- client := createConnection()
- request := iggcon.DeletePartitionRequest{
- StreamId: iggcon.NewIdentifier(int(createRandomUInt32())),
- TopicId: iggcon.NewIdentifier(int(createRandomUInt32())),
- PartitionsCount: 10,
- }
- err := client.DeletePartition(request)
+ client := createClient()
+ err := client.DeletePartitions(
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ 10,
+ )
itShouldReturnUnauthenticatedError(err)
})
diff --git a/bdd/go/tests/tcp_test/partitions_steps.go b/bdd/go/tests/tcp_test/partitions_steps.go
index 4af1fbc0..82045fb2 100644
--- a/bdd/go/tests/tcp_test/partitions_steps.go
+++ b/bdd/go/tests/tcp_test/partitions_steps.go
@@ -18,14 +18,14 @@
package tcp_test
import (
- "github.com/apache/iggy/foreign/go"
iggcon "github.com/apache/iggy/foreign/go/contracts"
+ "github.com/apache/iggy/foreign/go/iggycli"
. "github.com/onsi/ginkgo/v2"
. "github.com/onsi/gomega"
)
-func itShouldHaveExpectedNumberOfPartitions(streamId int, topicId int, expectedPartitions int, client iggy.MessageStream) {
- topic, err := client.GetTopicById(iggcon.NewIdentifier(streamId), iggcon.NewIdentifier(topicId))
+func itShouldHaveExpectedNumberOfPartitions(streamId int, topicId int, expectedPartitions int, client iggycli.Client) {
+ topic, err := client.GetTopic(iggcon.NewIdentifier(streamId), iggcon.NewIdentifier(topicId))
It("should have "+string(rune(expectedPartitions))+" partitions", func() {
Expect(topic).NotTo(BeNil())
diff --git a/bdd/go/tests/tcp_test/pat_feature_create.go b/bdd/go/tests/tcp_test/pat_feature_create.go
index 62c5fa9c..19455641 100644
--- a/bdd/go/tests/tcp_test/pat_feature_create.go
+++ b/bdd/go/tests/tcp_test/pat_feature_create.go
@@ -18,7 +18,6 @@
package tcp_test
import (
- iggcon "github.com/apache/iggy/foreign/go/contracts"
. "github.com/onsi/ginkgo/v2"
)
@@ -26,28 +25,19 @@ var _ = Describe("CREATE PAT:", func() {
When("User is logged in", func() {
Context("tries to create PAT with correct data", func() {
client := createAuthorizedConnection()
- request := iggcon.CreateAccessTokenRequest{
- Name: createRandomString(16),
- Expiry: 0,
- }
-
- response, err := client.CreateAccessToken(request)
+ name := createRandomString(16)
+ response, err := client.CreatePersonalAccessToken(name, 0)
itShouldNotReturnError(err)
- itShouldSuccessfullyCreateAccessToken(request.Name, client)
+ itShouldSuccessfullyCreateAccessToken(name, client)
itShouldBePossibleToLogInWithAccessToken(response.Token)
})
})
When("User is not logged in", func() {
Context("and tries to create PAT", func() {
- client := createConnection()
- request := iggcon.CreateAccessTokenRequest{
- Name: createRandomString(16),
- Expiry: 0,
- }
-
- _, err := client.CreateAccessToken(request)
+ client := createClient()
+ _, err := client.CreatePersonalAccessToken(createRandomString(16), 0)
itShouldReturnUnauthenticatedError(err)
})
})
diff --git a/bdd/go/tests/tcp_test/pat_feature_delete.go b/bdd/go/tests/tcp_test/pat_feature_delete.go
index fc3b5fc7..6aab6387 100644
--- a/bdd/go/tests/tcp_test/pat_feature_delete.go
+++ b/bdd/go/tests/tcp_test/pat_feature_delete.go
@@ -18,7 +18,6 @@
package tcp_test
import (
- iggcon "github.com/apache/iggy/foreign/go/contracts"
. "github.com/onsi/ginkgo/v2"
)
@@ -29,9 +28,7 @@ var _ = Describe("DELETE PAT:", func() {
name := createRandomString(16)
token := successfullyCreateAccessToken(name, client)
- err := client.DeleteAccessToken(iggcon.DeleteAccessTokenRequest{
- Name: name,
- })
+ err := client.DeletePersonalAccessToken(name)
itShouldNotReturnError(err)
itShouldSuccessfullyDeleteAccessToken(token, client)
@@ -40,10 +37,8 @@ var _ = Describe("DELETE PAT:", func() {
When("User is not logged in", func() {
Context("and tries to delete PAT", func() {
- client := createConnection()
- err := client.DeleteAccessToken(iggcon.DeleteAccessTokenRequest{
- Name: createRandomString(16),
- })
+ client := createClient()
+ err := client.DeletePersonalAccessToken(createRandomString(16))
itShouldReturnUnauthenticatedError(err)
})
})
diff --git a/bdd/go/tests/tcp_test/pat_feature_get_all.go b/bdd/go/tests/tcp_test/pat_feature_get_all.go
index 686edcfc..1dd5407e 100644
--- a/bdd/go/tests/tcp_test/pat_feature_get_all.go
+++ b/bdd/go/tests/tcp_test/pat_feature_get_all.go
@@ -28,7 +28,7 @@ var _ = Describe("GET PAT:", func() {
name := createRandomString(16)
successfullyCreateAccessToken(name, client)
- tokens, err := client.GetAccessTokens()
+ tokens, err := client.GetPersonalAccessTokens()
itShouldNotReturnError(err)
itShouldContainSpecificAccessToken(name, tokens)
@@ -37,8 +37,8 @@ var _ = Describe("GET PAT:", func() {
When("User is not logged in", func() {
Context("and tries to all get PAT's", func() {
- client := createConnection()
- _, err := client.GetAccessTokens()
+ client := createClient()
+ _, err := client.GetPersonalAccessTokens()
itShouldReturnUnauthenticatedError(err)
})
})
diff --git a/bdd/go/tests/tcp_test/pat_steps.go b/bdd/go/tests/tcp_test/pat_steps.go
index c9139057..65f4f986 100644
--- a/bdd/go/tests/tcp_test/pat_steps.go
+++ b/bdd/go/tests/tcp_test/pat_steps.go
@@ -18,21 +18,16 @@
package tcp_test
import (
- "github.com/apache/iggy/foreign/go"
iggcon "github.com/apache/iggy/foreign/go/contracts"
+ "github.com/apache/iggy/foreign/go/iggycli"
. "github.com/onsi/ginkgo/v2"
. "github.com/onsi/gomega"
)
// OPERATIONS
-func successfullyCreateAccessToken(name string, client iggy.MessageStream) string {
- request := iggcon.CreateAccessTokenRequest{
- Name: name,
- Expiry: 0,
- }
-
- result, err := client.CreateAccessToken(request)
+func successfullyCreateAccessToken(name string, client iggycli.Client) string {
+ result, err := client.CreatePersonalAccessToken(name, 0)
itShouldNotReturnError(err)
return result.Token
@@ -40,15 +35,15 @@ func successfullyCreateAccessToken(name string, client iggy.MessageStream) strin
// ASSERTIONS
-func itShouldSuccessfullyCreateAccessToken(name string, client iggy.MessageStream) {
- tokens, err := client.GetAccessTokens()
+func itShouldSuccessfullyCreateAccessToken(name string, client iggycli.Client) {
+ tokens, err := client.GetPersonalAccessTokens()
itShouldNotReturnError(err)
itShouldContainSpecificAccessToken(name, tokens)
}
-func itShouldSuccessfullyDeleteAccessToken(name string, client iggy.MessageStream) {
- tokens, err := client.GetAccessTokens()
+func itShouldSuccessfullyDeleteAccessToken(name string, client iggycli.Client) {
+ tokens, err := client.GetPersonalAccessTokens()
itShouldNotReturnError(err)
found := false
@@ -65,8 +60,8 @@ func itShouldSuccessfullyDeleteAccessToken(name string, client iggy.MessageStrea
}
func itShouldBePossibleToLogInWithAccessToken(token string) {
- ms := createConnection()
- userId, err := ms.LogInWithAccessToken(iggcon.LogInAccessTokenRequest{Token: token})
+ ms := createClient()
+ userId, err := ms.LoginWithPersonalAccessToken(token)
itShouldNotReturnError(err)
It("should return userId", func() {
@@ -74,12 +69,12 @@ func itShouldBePossibleToLogInWithAccessToken(token string) {
})
}
-func itShouldContainSpecificAccessToken(name string, tokens []iggcon.AccessTokenResponse) {
+func itShouldContainSpecificAccessToken(name string, tokens []iggcon.PersonalAccessTokenInfo) {
It("should fetch at least one user", func() {
Expect(len(tokens)).NotTo(Equal(0))
})
- var token iggcon.AccessTokenResponse
+ var token iggcon.PersonalAccessTokenInfo
found := false
for _, s := range tokens {
diff --git a/bdd/go/tests/tcp_test/ping_feature.go b/bdd/go/tests/tcp_test/ping_feature.go
index 7a0f4bb9..38889a1f 100644
--- a/bdd/go/tests/tcp_test/ping_feature.go
+++ b/bdd/go/tests/tcp_test/ping_feature.go
@@ -33,7 +33,7 @@ var _ = Describe("PING FEATURE:", func() {
When("User is not logged in", func() {
Context("and tries to ping server", func() {
- client := createConnection()
+ client := createClient()
err := client.Ping()
itShouldNotReturnError(err)
diff --git a/bdd/go/tests/tcp_test/session_feature_login.go b/bdd/go/tests/tcp_test/session_feature_login.go
index 4102a24c..fba6b841 100644
--- a/bdd/go/tests/tcp_test/session_feature_login.go
+++ b/bdd/go/tests/tcp_test/session_feature_login.go
@@ -27,10 +27,7 @@ var _ = Describe("LOGIN FEATURE:", func() {
When("user is already logged in", func() {
Context("and tries to log with correct data", func() {
client := createAuthorizedConnection()
- user, err := client.LogIn(LogInRequest{
- Username: "iggy",
- Password: "iggy",
- })
+ user, err := client.LoginUser("iggy", "iggy")
itShouldNotReturnError(err)
itShouldReturnUserId(user, 1)
@@ -38,10 +35,7 @@ var _ = Describe("LOGIN FEATURE:", func() {
Context("and tries to log with invalid credentials", func() {
client := createAuthorizedConnection()
- user, err := client.LogIn(LogInRequest{
- Username: "incorrect",
- Password: "random",
- })
+ user, err := client.LoginUser("incorrect", "random")
itShouldReturnError(err)
itShouldNotReturnUser(user)
@@ -50,22 +44,16 @@ var _ = Describe("LOGIN FEATURE:", func() {
When("user is not logged in", func() {
Context("and tries to log with correct data", func() {
- client := createConnection()
- user, err := client.LogIn(LogInRequest{
- Username: "iggy",
- Password: "iggy",
- })
+ client := createClient()
+ user, err := client.LoginUser("iggy", "iggy")
itShouldNotReturnError(err)
itShouldReturnUserId(user, 1)
})
Context("and tries to log with invalid credentials", func() {
- client := createConnection()
- user, err := client.LogIn(LogInRequest{
- Username: "incorrect",
- Password: "random",
- })
+ client := createClient()
+ user, err := client.LoginUser("incorrect", "random")
itShouldReturnError(err)
itShouldNotReturnUser(user)
@@ -73,13 +61,13 @@ var _ = Describe("LOGIN FEATURE:", func() {
})
})
-func itShouldReturnUserId(user *LogInResponse, id uint32) {
+func itShouldReturnUserId(user *IdentityInfo, id uint32) {
It("should return user id", func() {
Expect(user.UserId).To(Equal(id))
})
}
-func itShouldNotReturnUser(user *LogInResponse) {
+func itShouldNotReturnUser(user *IdentityInfo) {
It("should return user id", func() {
Expect(user).To(BeNil())
})
diff --git a/bdd/go/tests/tcp_test/session_feature_logout.go b/bdd/go/tests/tcp_test/session_feature_logout.go
index b099dbba..93d49bb7 100644
--- a/bdd/go/tests/tcp_test/session_feature_logout.go
+++ b/bdd/go/tests/tcp_test/session_feature_logout.go
@@ -25,7 +25,7 @@ var _ = Describe("LOGOUT FEATURE:", func() {
When("User is logged in", func() {
Context("and tries to log out", func() {
client := createAuthorizedConnection()
- err := client.LogOut()
+ err := client.LogoutUser()
itShouldNotReturnError(err)
})
@@ -33,8 +33,8 @@ var _ = Describe("LOGOUT FEATURE:", func() {
When("User is not logged in", func() {
Context("and tries to log out", func() {
- client := createConnection()
- err := client.LogOut()
+ client := createClient()
+ err := client.LogoutUser()
itShouldReturnUnauthenticatedError(err)
})
diff --git a/bdd/go/tests/tcp_test/stream_feature_create.go b/bdd/go/tests/tcp_test/stream_feature_create.go
index d84f12e8..1029667c 100644
--- a/bdd/go/tests/tcp_test/stream_feature_create.go
+++ b/bdd/go/tests/tcp_test/stream_feature_create.go
@@ -18,7 +18,6 @@
package tcp_test
import (
- iggcon "github.com/apache/iggy/foreign/go/contracts"
. "github.com/onsi/ginkgo/v2"
)
@@ -26,72 +25,55 @@ var _ = Describe("CREATE STREAM:", func() {
When("User is logged in", func() {
Context("and tries to create stream with unique name and id", func() {
client := createAuthorizedConnection()
- streamId := int(createRandomUInt32())
+ streamId := createRandomUInt32()
name := createRandomString(32)
- err := client.CreateStream(iggcon.CreateStreamRequest{
- StreamId: streamId,
- Name: name,
- })
- defer deleteStreamAfterTests(streamId, client)
+ _, err := client.CreateStream(name, &streamId)
+ defer deleteStreamAfterTests(int(streamId), client)
itShouldNotReturnError(err)
- itShouldSuccessfullyCreateStream(streamId, name, client)
+ itShouldSuccessfullyCreateStream(int(streamId), name, client)
})
Context("and tries to create stream with duplicate stream name", func() {
client := createAuthorizedConnection()
- streamId := int(createRandomUInt32())
+ streamId := createRandomUInt32()
name := createRandomString(32)
- err := client.CreateStream(iggcon.CreateStreamRequest{
- StreamId: streamId,
- Name: name,
- })
- defer deleteStreamAfterTests(streamId, client)
+ _, err := client.CreateStream(name, &streamId)
+ defer deleteStreamAfterTests(int(streamId), client)
itShouldNotReturnError(err)
- itShouldSuccessfullyCreateStream(streamId, name, client)
+ itShouldSuccessfullyCreateStream(int(streamId), name, client)
- err = client.CreateStream(iggcon.CreateStreamRequest{
- StreamId: int(createRandomUInt32()),
- Name: name,
- })
+ anotherStreamId := createRandomUInt32()
+ _, err = client.CreateStream(name, &anotherStreamId)
itShouldReturnSpecificError(err, "stream_name_already_exists")
})
Context("and tries to create stream with duplicate stream id", func() {
client := createAuthorizedConnection()
- streamId := int(createRandomUInt32())
+ streamId := createRandomUInt32()
name := createRandomString(32)
- err := client.CreateStream(iggcon.CreateStreamRequest{
- StreamId: streamId,
- Name: name,
- })
- defer deleteStreamAfterTests(streamId, client)
+ _, err := client.CreateStream(name, &streamId)
+ defer deleteStreamAfterTests(int(streamId), client)
itShouldNotReturnError(err)
- itShouldSuccessfullyCreateStream(streamId, name, client)
+ itShouldSuccessfullyCreateStream(int(streamId), name, client)
- err = client.CreateStream(iggcon.CreateStreamRequest{
- StreamId: streamId,
- Name: createRandomString(32),
- })
+ _, err = client.CreateStream(createRandomString(32), &streamId)
itShouldReturnSpecificError(err, "stream_id_already_exists")
})
Context("and tries to create stream name that's over 255 characters", func() {
client := createAuthorizedConnection()
- streamId := int(createRandomUInt32())
+ streamId := createRandomUInt32()
name := createRandomString(256)
- err := client.CreateStream(iggcon.CreateStreamRequest{
- StreamId: streamId,
- Name: name,
- })
+ _, err := client.CreateStream(name, &streamId)
itShouldReturnSpecificError(err, "stream_name_too_long")
})
@@ -99,11 +81,9 @@ var _ = Describe("CREATE STREAM:", func() {
When("User is not logged in", func() {
Context("and tries to create stream", func() {
- client := createConnection()
- err := client.CreateStream(iggcon.CreateStreamRequest{
- StreamId: int(createRandomUInt32()),
- Name: createRandomString(32),
- })
+ client := createClient()
+ streamId := createRandomUInt32()
+ _, err := client.CreateStream(createRandomString(32), &streamId)
itShouldReturnUnauthenticatedError(err)
})
diff --git a/bdd/go/tests/tcp_test/stream_feature_delete.go b/bdd/go/tests/tcp_test/stream_feature_delete.go
index 7166e773..45b7b480 100644
--- a/bdd/go/tests/tcp_test/stream_feature_delete.go
+++ b/bdd/go/tests/tcp_test/stream_feature_delete.go
@@ -46,7 +46,7 @@ var _ = Describe("DELETE STREAM:", func() {
When("User is not logged in", func() {
Context("and tries to delete stream", func() {
- client := createConnection()
+ client := createClient()
err := client.DeleteStream(iggcon.NewIdentifier(int(createRandomUInt32())))
itShouldReturnUnauthenticatedError(err)
diff --git a/bdd/go/tests/tcp_test/stream_feature_get_all.go b/bdd/go/tests/tcp_test/stream_feature_get_all.go
index 7f692dce..02eb24b0 100644
--- a/bdd/go/tests/tcp_test/stream_feature_get_all.go
+++ b/bdd/go/tests/tcp_test/stream_feature_get_all.go
@@ -37,7 +37,7 @@ var _ = Describe("GET ALL STREAMS:", func() {
When("User is not logged in", func() {
Context("and tries to get all streams", func() {
- client := createConnection()
+ client := createClient()
_, err := client.GetStreams()
itShouldReturnUnauthenticatedError(err)
diff --git a/bdd/go/tests/tcp_test/stream_feature_get_by_id.go b/bdd/go/tests/tcp_test/stream_feature_get_by_id.go
index 3406144e..d376418d 100644
--- a/bdd/go/tests/tcp_test/stream_feature_get_by_id.go
+++ b/bdd/go/tests/tcp_test/stream_feature_get_by_id.go
@@ -23,13 +23,13 @@ import (
)
var _ = Describe("GET STREAM BY ID:", func() {
- prefix := "GetStreamById"
+ prefix := "GetStream"
When("User is logged in", func() {
Context("and tries to get existing stream", func() {
client := createAuthorizedConnection()
streamId, name := successfullyCreateStream(prefix, client)
defer deleteStreamAfterTests(streamId, client)
- stream, err := client.GetStreamById(iggcon.GetStreamRequest{StreamID: iggcon.NewIdentifier(streamId)})
+ stream, err := client.GetStream(iggcon.NewIdentifier(streamId))
itShouldNotReturnError(err)
itShouldReturnSpecificStream(streamId, name, *stream)
@@ -39,7 +39,7 @@ var _ = Describe("GET STREAM BY ID:", func() {
client := createAuthorizedConnection()
streamId := int(createRandomUInt32())
- _, err := client.GetStreamById(iggcon.GetStreamRequest{StreamID: iggcon.NewIdentifier(streamId)})
+ _, err := client.GetStream(iggcon.NewIdentifier(streamId))
itShouldReturnSpecificError(err, "stream_id_not_found")
})
diff --git a/bdd/go/tests/tcp_test/stream_feature_update.go b/bdd/go/tests/tcp_test/stream_feature_update.go
index d368acbe..12585645 100644
--- a/bdd/go/tests/tcp_test/stream_feature_update.go
+++ b/bdd/go/tests/tcp_test/stream_feature_update.go
@@ -31,10 +31,7 @@ var _ = Describe("UPDATE STREAM:", func() {
defer deleteStreamAfterTests(streamId, client)
newName := createRandomString(128)
- err := client.UpdateStream(iggcon.UpdateStreamRequest{
- StreamId: iggcon.NewIdentifier(streamId),
- Name: newName,
- })
+ err := client.UpdateStream(iggcon.NewIdentifier(streamId), newName)
itShouldNotReturnError(err)
itShouldSuccessfullyUpdateStream(streamId, newName, client)
})
@@ -46,20 +43,14 @@ var _ = Describe("UPDATE STREAM:", func() {
defer deleteStreamAfterTests(stream1Id, client)
defer deleteStreamAfterTests(stream2Id, client)
- err := client.UpdateStream(iggcon.UpdateStreamRequest{
- StreamId: iggcon.NewIdentifier(stream2Id),
- Name: stream1Name,
- })
+ err := client.UpdateStream(iggcon.NewIdentifier(stream2Id), stream1Name)
itShouldReturnSpecificError(err, "stream_name_already_exists")
})
Context("and tries to update non-existing stream", func() {
client := createAuthorizedConnection()
- err := client.UpdateStream(iggcon.UpdateStreamRequest{
- StreamId: iggcon.NewIdentifier(int(createRandomUInt32())),
- Name: createRandomString(128),
- })
+ err := client.UpdateStream(iggcon.NewIdentifier(int(createRandomUInt32())), createRandomString(128))
itShouldReturnSpecificError(err, "stream_id_not_found")
})
@@ -69,10 +60,7 @@ var _ = Describe("UPDATE STREAM:", func() {
streamId, _ := successfullyCreateStream(prefix, client)
defer deleteStreamAfterTests(streamId, createAuthorizedConnection())
- err := client.UpdateStream(iggcon.UpdateStreamRequest{
- StreamId: iggcon.NewIdentifier(streamId),
- Name: createRandomString(256),
- })
+ err := client.UpdateStream(iggcon.NewIdentifier(streamId), createRandomString(256))
itShouldReturnSpecificError(err, "stream_name_too_long")
})
@@ -80,11 +68,8 @@ var _ = Describe("UPDATE STREAM:", func() {
When("User is not logged in", func() {
Context("and tries to update stream", func() {
- client := createConnection()
- err := client.UpdateStream(iggcon.UpdateStreamRequest{
- StreamId: iggcon.NewIdentifier(int(createRandomUInt32())),
- Name: createRandomString(128),
- })
+ client := createClient()
+ err := client.UpdateStream(iggcon.NewIdentifier(int(createRandomUInt32())), createRandomString(128))
itShouldReturnUnauthenticatedError(err)
})
diff --git a/bdd/go/tests/tcp_test/stream_steps.go b/bdd/go/tests/tcp_test/stream_steps.go
index 913932e2..cc76f532 100644
--- a/bdd/go/tests/tcp_test/stream_steps.go
+++ b/bdd/go/tests/tcp_test/stream_steps.go
@@ -20,32 +20,29 @@ package tcp_test
import (
"strconv"
- "github.com/apache/iggy/foreign/go"
iggcon "github.com/apache/iggy/foreign/go/contracts"
ierror "github.com/apache/iggy/foreign/go/errors"
+ "github.com/apache/iggy/foreign/go/iggycli"
. "github.com/onsi/ginkgo/v2"
. "github.com/onsi/gomega"
)
//operations
-func successfullyCreateStream(prefix string, client iggy.MessageStream) (int, string) {
- streamId := int(createRandomUInt32())
+func successfullyCreateStream(prefix string, client iggycli.Client) (int, string) {
+ streamId := createRandomUInt32()
name := createRandomStringWithPrefix(prefix, 128)
- err := client.CreateStream(iggcon.CreateStreamRequest{
- StreamId: streamId,
- Name: name,
- })
+ _, err := client.CreateStream(name, &streamId)
itShouldNotReturnError(err)
- itShouldSuccessfullyCreateStream(streamId, name, client)
- return streamId, name
+ itShouldSuccessfullyCreateStream(int(streamId), name, client)
+ return int(streamId), name
}
//assertions
-func itShouldReturnSpecificStream(id int, name string, stream iggcon.StreamResponse) {
+func itShouldReturnSpecificStream(id int, name string, stream iggcon.StreamDetails) {
It("should fetch stream with id "+string(rune(id)), func() {
Expect(stream.Id).To(Equal(id))
})
@@ -55,12 +52,12 @@ func itShouldReturnSpecificStream(id int, name string, stream iggcon.StreamRespo
})
}
-func itShouldContainSpecificStream(id int, name string, streams []iggcon.StreamResponse) {
+func itShouldContainSpecificStream(id int, name string, streams []iggcon.Stream) {
It("should fetch at least one stream", func() {
Expect(len(streams)).NotTo(Equal(0))
})
- var stream iggcon.StreamResponse
+ var stream iggcon.Stream
found := false
for _, s := range streams {
@@ -82,8 +79,8 @@ func itShouldContainSpecificStream(id int, name string, streams []iggcon.StreamR
})
}
-func itShouldSuccessfullyCreateStream(id int, expectedName string, client iggy.MessageStream) {
- stream, err := client.GetStreamById(iggcon.GetStreamRequest{StreamID: iggcon.NewIdentifier(id)})
+func itShouldSuccessfullyCreateStream(id int, expectedName string, client iggycli.Client) {
+ stream, err := client.GetStream(iggcon.NewIdentifier(id))
itShouldNotReturnError(err)
It("should create stream with id "+string(rune(id)), func() {
@@ -95,8 +92,8 @@ func itShouldSuccessfullyCreateStream(id int, expectedName string, client iggy.M
})
}
-func itShouldSuccessfullyUpdateStream(id int, expectedName string, client iggy.MessageStream) {
- stream, err := client.GetStreamById(iggcon.GetStreamRequest{StreamID: iggcon.NewIdentifier(id)})
+func itShouldSuccessfullyUpdateStream(id int, expectedName string, client iggycli.Client) {
+ stream, err := client.GetStream(iggcon.NewIdentifier(id))
itShouldNotReturnError(err)
It("should update stream with id "+string(rune(id)), func() {
@@ -108,8 +105,8 @@ func itShouldSuccessfullyUpdateStream(id int, expectedName string, client iggy.M
})
}
-func itShouldSuccessfullyDeleteStream(id int, client iggy.MessageStream) {
- stream, err := client.GetStreamById(iggcon.GetStreamRequest{StreamID: iggcon.NewIdentifier(id)})
+func itShouldSuccessfullyDeleteStream(id int, client iggycli.Client) {
+ stream, err := client.GetStream(iggcon.NewIdentifier(id))
itShouldReturnSpecificIggyError(err, ierror.StreamIdNotFound)
It("should not return stream", func() {
@@ -117,6 +114,6 @@ func itShouldSuccessfullyDeleteStream(id int, client iggy.MessageStream) {
})
}
-func deleteStreamAfterTests(streamId int, client iggy.MessageStream) {
+func deleteStreamAfterTests(streamId int, client iggycli.Client) {
_ = client.DeleteStream(iggcon.NewIdentifier(streamId))
}
diff --git a/bdd/go/tests/tcp_test/test_helpers.go b/bdd/go/tests/tcp_test/test_helpers.go
index 68d4f4e3..f30df5e6 100644
--- a/bdd/go/tests/tcp_test/test_helpers.go
+++ b/bdd/go/tests/tcp_test/test_helpers.go
@@ -23,38 +23,33 @@ import (
"strings"
"time"
- . "github.com/apache/iggy/foreign/go"
- . "github.com/apache/iggy/foreign/go/contracts"
+ "github.com/apache/iggy/foreign/go/iggycli"
+ "github.com/apache/iggy/foreign/go/tcp"
)
-func createAuthorizedConnection() MessageStream {
- ms := createConnection()
- _, err := ms.LogIn(LogInRequest{
- Username: "iggy",
- Password: "iggy",
- })
+func createAuthorizedConnection() iggycli.Client {
+ cli := createClient()
+ _, err := cli.LoginUser("iggy", "iggy")
if err != nil {
panic(err)
}
- return ms
+ return cli
}
-func createConnection() MessageStream {
+func createClient() iggycli.Client {
addr := os.Getenv("IGGY_TCP_ADDRESS")
if addr == "" {
addr = "127.0.0.1:8090"
}
- factory := &IggyClientFactory{}
- config := IggyConfiguration{
- BaseAddress: addr,
- Protocol: Tcp,
- }
-
- ms, err := factory.CreateMessageStream(config)
+ cli, err := iggycli.NewIggyClient(
+ iggycli.WithTcp(
+ tcp.WithServerAddress(addr),
+ ),
+ )
if err != nil {
panic(err)
}
- return ms
+ return cli
}
func createRandomUInt32() uint32 {
diff --git a/bdd/go/tests/tcp_test/topic_feature_create.go b/bdd/go/tests/tcp_test/topic_feature_create.go
index 623ff693..7015c17e 100644
--- a/bdd/go/tests/tcp_test/topic_feature_create.go
+++ b/bdd/go/tests/tcp_test/topic_feature_create.go
@@ -30,39 +30,39 @@ var _ = Describe("CREATE TOPIC:", func() {
Context("and tries to create topic unique name and id", func() {
client := createAuthorizedConnection()
streamId, _ := successfullyCreateStream(prefix, client)
+ topicId := 1
+ replicationFactor := uint8(1)
+ name := createRandomString(32)
defer deleteStreamAfterTests(streamId, client)
-
- request := iggcon.CreateTopicRequest{
- TopicId: 1,
- StreamId: iggcon.NewIdentifier(streamId),
- Name: createRandomString(32),
- MessageExpiry: 1000,
- PartitionsCount: 2,
- CompressionAlgorithm: 1,
- MaxTopicSize: math.MaxUint64,
- ReplicationFactor: 1,
- }
- err := client.CreateTopic(request)
+ _, err := client.CreateTopic(
+ iggcon.NewIdentifier(streamId),
+ name,
+ 2,
+ 1,
+ 1000,
+ math.MaxUint64,
+ &replicationFactor,
+ &topicId)
itShouldNotReturnError(err)
- itShouldSuccessfullyCreateTopic(streamId, request.TopicId, request.Name, client)
+ itShouldSuccessfullyCreateTopic(streamId, topicId, name, client)
})
Context("and tries to create topic for a non existing stream", func() {
client := createAuthorizedConnection()
streamId := int(createRandomUInt32())
-
- request := iggcon.CreateTopicRequest{
- TopicId: 1,
- StreamId: iggcon.NewIdentifier(streamId),
- Name: createRandomString(32),
- MessageExpiry: 1000,
- PartitionsCount: 2,
- CompressionAlgorithm: 1,
- MaxTopicSize: math.MaxUint64,
- ReplicationFactor: 1,
- }
- err := client.CreateTopic(request)
+ topicId := 1
+ replicationFactor := uint8(1)
+ name := createRandomString(32)
+ _, err := client.CreateTopic(
+ iggcon.NewIdentifier(streamId),
+ name,
+ 2,
+ 1,
+ 1000,
+ math.MaxUint64,
+ &replicationFactor,
+ &topicId)
itShouldReturnSpecificError(err, "stream_id_not_found")
})
@@ -73,17 +73,17 @@ var _ = Describe("CREATE TOPIC:", func() {
defer deleteStreamAfterTests(streamId, client)
_, name := successfullyCreateTopic(streamId, client)
- request := iggcon.CreateTopicRequest{
- TopicId: int(createRandomUInt32()),
- StreamId: iggcon.NewIdentifier(streamId),
- Name: name,
- MessageExpiry: 0,
- PartitionsCount: 2,
- CompressionAlgorithm: 1,
- MaxTopicSize: math.MaxUint64,
- ReplicationFactor: 1,
- }
- err := client.CreateTopic(request)
+ replicationFactor := uint8(1)
+ topicId := int(createRandomUInt32())
+ _, err := client.CreateTopic(
+ iggcon.NewIdentifier(streamId),
+ name,
+ 2,
+ 1,
+ 0,
+ math.MaxUint64,
+ &replicationFactor,
+ &topicId)
itShouldReturnSpecificError(err, "topic_name_already_exists")
})
@@ -93,17 +93,16 @@ var _ = Describe("CREATE TOPIC:", func() {
defer deleteStreamAfterTests(streamId, client)
topicId, _ := successfullyCreateTopic(streamId, client)
- request := iggcon.CreateTopicRequest{
- TopicId: topicId,
- StreamId: iggcon.NewIdentifier(streamId),
- Name: createRandomString(32),
- MessageExpiry: 0,
- PartitionsCount: 2,
- CompressionAlgorithm: 1,
- MaxTopicSize: math.MaxUint64,
- ReplicationFactor: 1,
- }
- err := client.CreateTopic(request)
+ replicationFactor := uint8(1)
+ _, err := client.CreateTopic(
+ iggcon.NewIdentifier(streamId),
+ createRandomString(32),
+ 2,
+ 1,
+ 0,
+ math.MaxUint64,
+ &replicationFactor,
+ &topicId)
itShouldReturnSpecificError(err, "topic_id_already_exists")
})
@@ -112,17 +111,17 @@ var _ = Describe("CREATE TOPIC:", func() {
streamId, _ := successfullyCreateStream(prefix, client)
defer deleteStreamAfterTests(streamId, createAuthorizedConnection())
- request := iggcon.CreateTopicRequest{
- TopicId: int(createRandomUInt32()),
- StreamId: iggcon.NewIdentifier(streamId),
- Name: createRandomString(256),
- MessageExpiry: 0,
- PartitionsCount: 2,
- CompressionAlgorithm: 1,
- MaxTopicSize: math.MaxUint64,
- ReplicationFactor: 1,
- }
- err := client.CreateTopic(request)
+ replicationFactor := uint8(1)
+ topicId := int(createRandomUInt32())
+ _, err := client.CreateTopic(
+ iggcon.NewIdentifier(streamId),
+ createRandomString(256),
+ 2,
+ 1,
+ 0,
+ math.MaxUint64,
+ &replicationFactor,
+ &topicId)
itShouldReturnSpecificError(err, "topic_name_too_long")
})
@@ -130,17 +129,18 @@ var _ = Describe("CREATE TOPIC:", func() {
When("User is not logged in", func() {
Context("and tries to create topic", func() {
- client := createConnection()
- err := client.CreateTopic(iggcon.CreateTopicRequest{
- TopicId: 1,
- StreamId: iggcon.NewIdentifier(10),
- Name: "name",
- MessageExpiry: 0,
- PartitionsCount: 2,
- CompressionAlgorithm: 1,
- MaxTopicSize: math.MaxUint64,
- ReplicationFactor: 1,
- })
+ client := createClient()
+ replicationFactor := uint8(1)
+ topicId := 1
+ _, err := client.CreateTopic(
+ iggcon.NewIdentifier(10),
+ "name",
+ 2,
+ 1,
+ 0,
+ math.MaxUint64,
+ &replicationFactor,
+ &topicId)
itShouldReturnUnauthenticatedError(err)
})
diff --git a/bdd/go/tests/tcp_test/topic_feature_delete.go b/bdd/go/tests/tcp_test/topic_feature_delete.go
index 2f242d38..22ca74b0 100644
--- a/bdd/go/tests/tcp_test/topic_feature_delete.go
+++ b/bdd/go/tests/tcp_test/topic_feature_delete.go
@@ -61,7 +61,7 @@ var _ = Describe("DELETE TOPIC:", func() {
When("User is not logged in", func() {
Context("and tries to delete topic", func() {
- client := createConnection()
+ client := createClient()
err := client.DeleteTopic(iggcon.NewIdentifier(int(createRandomUInt32())), iggcon.NewIdentifier(int(createRandomUInt32())))
itShouldReturnUnauthenticatedError(err)
diff --git a/bdd/go/tests/tcp_test/topic_feature_get_all.go b/bdd/go/tests/tcp_test/topic_feature_get_all.go
index 063182d7..294d338e 100644
--- a/bdd/go/tests/tcp_test/topic_feature_get_all.go
+++ b/bdd/go/tests/tcp_test/topic_feature_get_all.go
@@ -39,7 +39,7 @@ var _ = Describe("GET ALL TOPICS:", func() {
When("User is not logged in", func() {
Context("and tries to get all topics", func() {
- client := createConnection()
+ client := createClient()
_, err := client.GetTopics(iggcon.NewIdentifier(int(createRandomUInt32())))
itShouldReturnUnauthenticatedError(err)
diff --git a/bdd/go/tests/tcp_test/topic_feature_get_by_id.go b/bdd/go/tests/tcp_test/topic_feature_get_by_id.go
index ad0eba79..2371e889 100644
--- a/bdd/go/tests/tcp_test/topic_feature_get_by_id.go
+++ b/bdd/go/tests/tcp_test/topic_feature_get_by_id.go
@@ -24,14 +24,14 @@ import (
)
var _ = Describe("GET TOPIC BY ID:", func() {
- prefix := "GetTopicById"
+ prefix := "GetTopic"
When("User is logged in", func() {
Context("and tries to get existing topic", func() {
client := createAuthorizedConnection()
streamId, _ := successfullyCreateStream(prefix, client)
defer deleteStreamAfterTests(streamId, client)
topicId, name := successfullyCreateTopic(streamId, client)
- topic, err := client.GetTopicById(iggcon.NewIdentifier(streamId), iggcon.NewIdentifier(topicId))
+ topic, err := client.GetTopic(iggcon.NewIdentifier(streamId), iggcon.NewIdentifier(topicId))
itShouldNotReturnError(err)
itShouldReturnSpecificTopic(topicId, name, *topic)
@@ -41,7 +41,7 @@ var _ = Describe("GET TOPIC BY ID:", func() {
client := createAuthorizedConnection()
streamId := int(createRandomUInt32())
- _, err := client.GetTopicById(iggcon.NewIdentifier(streamId), iggcon.NewIdentifier(int(createRandomUInt32())))
+ _, err := client.GetTopic(iggcon.NewIdentifier(streamId), iggcon.NewIdentifier(int(createRandomUInt32())))
itShouldReturnSpecificIggyError(err, ierror.TopicIdNotFound)
})
@@ -51,7 +51,7 @@ var _ = Describe("GET TOPIC BY ID:", func() {
streamId, _ := successfullyCreateStream(prefix, client)
defer deleteStreamAfterTests(streamId, client)
- _, err := client.GetTopicById(iggcon.NewIdentifier(streamId), iggcon.NewIdentifier(int(createRandomUInt32())))
+ _, err := client.GetTopic(iggcon.NewIdentifier(streamId), iggcon.NewIdentifier(int(createRandomUInt32())))
itShouldReturnSpecificIggyError(err, ierror.TopicIdNotFound)
})
diff --git a/bdd/go/tests/tcp_test/topic_feature_update.go b/bdd/go/tests/tcp_test/topic_feature_update.go
index 3235a4b8..5ac97ccb 100644
--- a/bdd/go/tests/tcp_test/topic_feature_update.go
+++ b/bdd/go/tests/tcp_test/topic_feature_update.go
@@ -33,16 +33,14 @@ var _ = Describe("UPDATE TOPIC:", func() {
defer deleteStreamAfterTests(streamId, client)
topicId, _ := successfullyCreateTopic(streamId, client)
newName := createRandomString(128)
- request := iggcon.UpdateTopicRequest{
- TopicId: iggcon.NewIdentifier(topicId),
- StreamId: iggcon.NewIdentifier(streamId),
- Name: newName,
- MessageExpiry: 1,
- CompressionAlgorithm: 1,
- MaxTopicSize: math.MaxUint64,
- ReplicationFactor: 1,
- }
- err := client.UpdateTopic(request)
+ replicationFactor := uint8(1)
+ err := client.UpdateTopic(iggcon.NewIdentifier(streamId),
+ iggcon.NewIdentifier(topicId),
+ newName,
+ 1,
+ 1,
+ math.MaxUint64,
+ &replicationFactor)
itShouldNotReturnError(err)
itShouldSuccessfullyUpdateTopic(streamId, topicId, newName, client)
})
@@ -53,17 +51,14 @@ var _ = Describe("UPDATE TOPIC:", func() {
defer deleteStreamAfterTests(streamId, client)
_, topic1Name := successfullyCreateTopic(streamId, client)
topic2Id, _ := successfullyCreateTopic(streamId, client)
-
- request := iggcon.UpdateTopicRequest{
- TopicId: iggcon.NewIdentifier(topic2Id),
- StreamId: iggcon.NewIdentifier(streamId),
- Name: topic1Name,
- MessageExpiry: 0,
- CompressionAlgorithm: 1,
- MaxTopicSize: math.MaxUint64,
- ReplicationFactor: 1,
- }
- err := client.UpdateTopic(request)
+ replicationFactor := uint8(1)
+ err := client.UpdateTopic(iggcon.NewIdentifier(streamId),
+ iggcon.NewIdentifier(topic2Id),
+ topic1Name,
+ 1,
+ 0,
+ math.MaxUint64,
+ &replicationFactor)
itShouldReturnSpecificError(err, "topic_name_already_exists")
})
@@ -72,16 +67,15 @@ var _ = Describe("UPDATE TOPIC:", func() {
client := createAuthorizedConnection()
streamId := int(createRandomUInt32())
topicId := int(createRandomUInt32())
- request := iggcon.UpdateTopicRequest{
- TopicId: iggcon.NewIdentifier(topicId),
- StreamId: iggcon.NewIdentifier(streamId),
- Name: createRandomString(128),
- MessageExpiry: 0,
- CompressionAlgorithm: 1,
- MaxTopicSize: math.MaxUint64,
- ReplicationFactor: 1,
- }
- err := client.UpdateTopic(request)
+ replicationFactor := uint8(1)
+ err := client.UpdateTopic(
+ iggcon.NewIdentifier(streamId),
+ iggcon.NewIdentifier(topicId),
+ createRandomString(128),
+ 1,
+ 0,
+ math.MaxUint64,
+ &replicationFactor)
itShouldReturnSpecificError(err, "stream_id_not_found")
})
@@ -91,16 +85,15 @@ var _ = Describe("UPDATE TOPIC:", func() {
streamId, _ := successfullyCreateStream(prefix, client)
defer deleteStreamAfterTests(streamId, createAuthorizedConnection())
topicId := int(createRandomUInt32())
- request := iggcon.UpdateTopicRequest{
- TopicId: iggcon.NewIdentifier(topicId),
- StreamId: iggcon.NewIdentifier(streamId),
- Name: createRandomString(128),
- MessageExpiry: 0,
- CompressionAlgorithm: 1,
- MaxTopicSize: math.MaxUint64,
- ReplicationFactor: 1,
- }
- err := client.UpdateTopic(request)
+ replicationFactor := uint8(1)
+ err := client.UpdateTopic(
+ iggcon.NewIdentifier(streamId),
+ iggcon.NewIdentifier(topicId),
+ createRandomString(128),
+ 1,
+ 0,
+ math.MaxUint64,
+ &replicationFactor)
itShouldReturnSpecificError(err, "topic_id_not_found")
})
@@ -110,17 +103,15 @@ var _ = Describe("UPDATE TOPIC:", func() {
streamId, _ := successfullyCreateStream(prefix, client)
defer deleteStreamAfterTests(streamId, createAuthorizedConnection())
topicId, _ := successfullyCreateTopic(streamId, client)
- request := iggcon.UpdateTopicRequest{
- TopicId: iggcon.NewIdentifier(topicId),
- StreamId: iggcon.NewIdentifier(streamId),
- Name: createRandomString(256),
- MessageExpiry: 0,
- CompressionAlgorithm: 1,
- MaxTopicSize: math.MaxUint64,
- ReplicationFactor: 1,
- }
-
- err := client.UpdateTopic(request)
+ replicationFactor := uint8(1)
+ err := client.UpdateTopic(
+ iggcon.NewIdentifier(streamId),
+ iggcon.NewIdentifier(topicId),
+ createRandomString(256),
+ 1,
+ 0,
+ math.MaxUint64,
+ &replicationFactor)
itShouldReturnSpecificError(err, "topic_name_too_long")
})
@@ -128,11 +119,8 @@ var _ = Describe("UPDATE TOPIC:", func() {
When("User is not logged in", func() {
Context("and tries to update stream", func() {
- client := createConnection()
- err := client.UpdateStream(iggcon.UpdateStreamRequest{
- StreamId: iggcon.NewIdentifier(int(createRandomUInt32())),
- Name: createRandomString(128),
- })
+ client := createClient()
+ err := client.UpdateStream(iggcon.NewIdentifier(int(createRandomUInt32())), createRandomString(128))
itShouldReturnUnauthenticatedError(err)
})
diff --git a/bdd/go/tests/tcp_test/topic_steps.go b/bdd/go/tests/tcp_test/topic_steps.go
index ebb98b32..f956c3b2 100644
--- a/bdd/go/tests/tcp_test/topic_steps.go
+++ b/bdd/go/tests/tcp_test/topic_steps.go
@@ -21,36 +21,37 @@ import (
"math"
"strconv"
- "github.com/apache/iggy/foreign/go"
iggcon "github.com/apache/iggy/foreign/go/contracts"
ierror "github.com/apache/iggy/foreign/go/errors"
+ "github.com/apache/iggy/foreign/go/iggycli"
. "github.com/onsi/ginkgo/v2"
. "github.com/onsi/gomega"
)
//operations
-func successfullyCreateTopic(streamId int, client iggy.MessageStream) (int, string) {
- request := iggcon.CreateTopicRequest{
- TopicId: int(createRandomUInt32()),
- StreamId: iggcon.NewIdentifier(streamId),
- Name: createRandomString(128),
- MessageExpiry: 0,
- PartitionsCount: 2,
- CompressionAlgorithm: 1,
- MaxTopicSize: math.MaxUint64,
- ReplicationFactor: 1,
- }
- err := client.CreateTopic(request)
-
- itShouldSuccessfullyCreateTopic(streamId, request.TopicId, request.Name, client)
+func successfullyCreateTopic(streamId int, client iggycli.Client) (int, string) {
+ topicId := int(createRandomUInt32())
+ replicationFactor := uint8(1)
+ name := createRandomString(128)
+ _, err := client.CreateTopic(
+ iggcon.NewIdentifier(streamId),
+ name,
+ 2,
+ 1,
+ 0,
+ math.MaxUint64,
+ &replicationFactor,
+ &topicId)
+
+ itShouldSuccessfullyCreateTopic(streamId, topicId, name, client)
itShouldNotReturnError(err)
- return request.TopicId, request.Name
+ return topicId, name
}
//assertions
-func itShouldReturnSpecificTopic(id int, name string, topic iggcon.TopicResponse) {
+func itShouldReturnSpecificTopic(id int, name string, topic iggcon.TopicDetails) {
It("should fetch topic with id "+string(rune(id)), func() {
Expect(topic.Id).To(Equal(id))
})
@@ -60,12 +61,12 @@ func itShouldReturnSpecificTopic(id int, name string, topic iggcon.TopicResponse
})
}
-func itShouldContainSpecificTopic(id int, name string, topics []iggcon.TopicResponse) {
+func itShouldContainSpecificTopic(id int, name string, topics []iggcon.Topic) {
It("should fetch at least one topic", func() {
Expect(len(topics)).NotTo(Equal(0))
})
- var topic iggcon.TopicResponse
+ var topic iggcon.Topic
found := false
for _, s := range topics {
@@ -87,8 +88,8 @@ func itShouldContainSpecificTopic(id int, name string, topics []iggcon.TopicResp
})
}
-func itShouldSuccessfullyCreateTopic(streamId int, topicId int, expectedName string, client iggy.MessageStream) {
- topic, err := client.GetTopicById(iggcon.NewIdentifier(streamId), iggcon.NewIdentifier(topicId))
+func itShouldSuccessfullyCreateTopic(streamId int, topicId int, expectedName string, client iggycli.Client) {
+ topic, err := client.GetTopic(iggcon.NewIdentifier(streamId), iggcon.NewIdentifier(topicId))
It("should create topic with id "+string(rune(topicId)), func() {
Expect(topic).NotTo(BeNil())
Expect(topic.Id).To(Equal(topicId))
@@ -101,8 +102,8 @@ func itShouldSuccessfullyCreateTopic(streamId int, topicId int, expectedName str
itShouldNotReturnError(err)
}
-func itShouldSuccessfullyUpdateTopic(streamId int, topicId int, expectedName string, client iggy.MessageStream) {
- topic, err := client.GetTopicById(iggcon.NewIdentifier(streamId), iggcon.NewIdentifier(topicId))
+func itShouldSuccessfullyUpdateTopic(streamId int, topicId int, expectedName string, client iggycli.Client) {
+ topic, err := client.GetTopic(iggcon.NewIdentifier(streamId), iggcon.NewIdentifier(topicId))
It("should update topic with id "+string(rune(topicId)), func() {
Expect(topic).NotTo(BeNil())
@@ -116,8 +117,8 @@ func itShouldSuccessfullyUpdateTopic(streamId int, topicId int, expectedName str
itShouldNotReturnError(err)
}
-func itShouldSuccessfullyDeleteTopic(streamId int, topicId int, client iggy.MessageStream) {
- topic, err := client.GetTopicById(iggcon.NewIdentifier(streamId), iggcon.NewIdentifier(topicId))
+func itShouldSuccessfullyDeleteTopic(streamId int, topicId int, client iggycli.Client) {
+ topic, err := client.GetTopic(iggcon.NewIdentifier(streamId), iggcon.NewIdentifier(topicId))
itShouldReturnSpecificIggyError(err, ierror.TopicIdNotFound)
It("should not return topic", func() {
diff --git a/bdd/go/tests/tcp_test/users_feature_create.go b/bdd/go/tests/tcp_test/users_feature_create.go
index df658953..abecbe0f 100644
--- a/bdd/go/tests/tcp_test/users_feature_create.go
+++ b/bdd/go/tests/tcp_test/users_feature_create.go
@@ -27,11 +27,12 @@ var _ = Describe("CREATE USER:", func() {
Context("tries to create user with correct data", func() {
client := createAuthorizedConnection()
- request := iggcon.CreateUserRequest{
- Username: createRandomString(16),
- Password: createRandomString(16),
- Status: iggcon.Active,
- Permissions: &iggcon.Permissions{
+ username := createRandomString(16)
+ _, err := client.CreateUser(
+ username,
+ createRandomString(16),
+ iggcon.Active,
+ &iggcon.Permissions{
Global: iggcon.GlobalPermissions{
ManageServers: true,
ReadServers: true,
@@ -44,14 +45,11 @@ var _ = Describe("CREATE USER:", func() {
PollMessages: true,
SendMessages: true,
},
- },
- }
-
- err := client.CreateUser(request)
- defer deleteUserAfterTests(request.Username, client)
+ })
+ defer deleteUserAfterTests(username, client)
itShouldNotReturnError(err)
- itShouldSuccessfullyCreateUser(request.Username, client)
+ itShouldSuccessfullyCreateUser(username, client)
//itShouldBePossibleToLogInWithCredentials(request.Username, request.Password)
})
@@ -82,11 +80,12 @@ var _ = Describe("CREATE USER:", func() {
int(streamId): &streamPermissionRequest,
}
- request := iggcon.CreateUserRequest{
- Username: createRandomString(16),
- Password: createRandomString(16),
- Status: iggcon.Active,
- Permissions: &iggcon.Permissions{
+ username := createRandomString(16)
+ _, err := client.CreateUser(
+ username,
+ createRandomString(16),
+ iggcon.Active,
+ &iggcon.Permissions{
Global: iggcon.GlobalPermissions{
ManageServers: true,
ReadServers: true,
@@ -100,27 +99,25 @@ var _ = Describe("CREATE USER:", func() {
SendMessages: true,
},
Streams: userStreamPermissions,
- },
- }
-
- err := client.CreateUser(request)
- defer deleteUserAfterTests(request.Username, client)
+ })
+ defer deleteUserAfterTests(username, client)
defer deleteStreamAfterTests(streamId, client)
itShouldNotReturnError(err)
- itShouldSuccessfullyCreateUserWithPermissions(request.Username, client, userStreamPermissions)
+ itShouldSuccessfullyCreateUserWithPermissions(username, client, userStreamPermissions)
//itShouldBePossibleToLogInWithCredentials(request.Username, request.Password)
})
})
When("User is not logged in", func() {
Context("and tries to create user", func() {
- client := createConnection()
- request := iggcon.CreateUserRequest{
- Username: createRandomString(16),
- Password: createRandomString(16),
- Status: iggcon.Active,
- Permissions: &iggcon.Permissions{
+ client := createClient()
+
+ _, err := client.CreateUser(
+ createRandomString(16),
+ createRandomString(16),
+ iggcon.Active,
+ &iggcon.Permissions{
Global: iggcon.GlobalPermissions{
ManageServers: true,
ReadServers: true,
@@ -133,10 +130,8 @@ var _ = Describe("CREATE USER:", func() {
PollMessages: true,
SendMessages: true,
},
- },
- }
+ })
- err := client.CreateUser(request)
itShouldReturnUnauthenticatedError(err)
})
})
diff --git a/bdd/go/tests/tcp_test/users_feature_delete.go b/bdd/go/tests/tcp_test/users_feature_delete.go
index c320e95a..e362f70e 100644
--- a/bdd/go/tests/tcp_test/users_feature_delete.go
+++ b/bdd/go/tests/tcp_test/users_feature_delete.go
@@ -37,7 +37,7 @@ var _ = Describe("DELETE USER:", func() {
When("User is not logged in", func() {
Context("and tries to delete user", func() {
- client := createConnection()
+ client := createClient()
err := client.DeleteUser(iggcon.NewIdentifier(int(createRandomUInt32())))
itShouldReturnUnauthenticatedError(err)
})
diff --git a/bdd/go/tests/tcp_test/users_feature_get_all.go b/bdd/go/tests/tcp_test/users_feature_get_all.go
index c291c7fe..4c2b91c3 100644
--- a/bdd/go/tests/tcp_test/users_feature_get_all.go
+++ b/bdd/go/tests/tcp_test/users_feature_get_all.go
@@ -38,7 +38,7 @@ var _ = Describe("GET USER:", func() {
When("User is not logged in", func() {
Context("and tries to all get users", func() {
- client := createConnection()
+ client := createClient()
_, err := client.GetUsers()
itShouldReturnUnauthenticatedError(err)
})
diff --git a/bdd/go/tests/tcp_test/users_feature_get_by_id.go b/bdd/go/tests/tcp_test/users_feature_get_by_id.go
index f3f983cd..b1e4d966 100644
--- a/bdd/go/tests/tcp_test/users_feature_get_by_id.go
+++ b/bdd/go/tests/tcp_test/users_feature_get_by_id.go
@@ -33,7 +33,7 @@ var _ = Describe("GET USER:", func() {
user, err := client.GetUser(iggcon.NewIdentifier(int(userId)))
itShouldNotReturnError(err)
- itShouldReturnSpecificUser(name, *user)
+ itShouldReturnSpecificUser(name, user.UserInfo)
})
})
diff --git a/bdd/go/tests/tcp_test/users_feature_password.go b/bdd/go/tests/tcp_test/users_feature_password.go
index 1972980f..9706c992 100644
--- a/bdd/go/tests/tcp_test/users_feature_password.go
+++ b/bdd/go/tests/tcp_test/users_feature_password.go
@@ -26,11 +26,14 @@ var _ = Describe("CHANGE PASSWORD:", func() {
When("User is logged in", func() {
Context("tries to change password of existing user", func() {
client := createAuthorizedConnection()
- createRequest := iggcon.CreateUserRequest{
- Username: createRandomStringWithPrefix("ch_p_", 16),
- Password: "oldPassword",
- Status: iggcon.Active,
- Permissions: &iggcon.Permissions{
+
+ username := createRandomStringWithPrefix("ch_p_", 16)
+ password := "oldPassword"
+ _, err := client.CreateUser(
+ username,
+ password,
+ iggcon.Active,
+ &iggcon.Permissions{
Global: iggcon.GlobalPermissions{
ManageServers: true,
ReadServers: true,
@@ -43,18 +46,10 @@ var _ = Describe("CHANGE PASSWORD:", func() {
PollMessages: true,
SendMessages: true,
},
- },
- }
-
- err := client.CreateUser(createRequest)
- defer deleteUserAfterTests(createRequest.Username, client)
- request := iggcon.ChangePasswordRequest{
- UserID: iggcon.NewIdentifier(createRequest.Username),
- CurrentPassword: createRequest.Password,
- NewPassword: "newPassword",
- }
+ })
+ defer deleteUserAfterTests(username, client)
- err = client.ChangePassword(request)
+ err = client.ChangePassword(iggcon.NewIdentifier(username), password, "newPassword")
itShouldNotReturnError(err)
//itShouldBePossibleToLogInWithCredentials(createRequest.Username, request.NewPassword)
@@ -63,10 +58,11 @@ var _ = Describe("CHANGE PASSWORD:", func() {
When("User is not logged in", func() {
Context("and tries to change password", func() {
- client := createConnection()
- request := iggcon.UpdateUserPermissionsRequest{
- UserID: iggcon.NewIdentifier(int(createRandomUInt32())),
- Permissions: &iggcon.Permissions{
+ client := createClient()
+
+ err := client.UpdatePermissions(
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ &iggcon.Permissions{
Global: iggcon.GlobalPermissions{
ManageServers: false,
ReadServers: false,
@@ -79,10 +75,7 @@ var _ = Describe("CHANGE PASSWORD:", func() {
PollMessages: false,
SendMessages: false,
},
- },
- }
-
- err := client.UpdateUserPermissions(request)
+ })
itShouldReturnUnauthenticatedError(err)
})
})
diff --git a/bdd/go/tests/tcp_test/users_feature_permissions.go b/bdd/go/tests/tcp_test/users_feature_permissions.go
index 21e6cbab..e891bc45 100644
--- a/bdd/go/tests/tcp_test/users_feature_permissions.go
+++ b/bdd/go/tests/tcp_test/users_feature_permissions.go
@@ -28,9 +28,10 @@ var _ = Describe("UPDATE USER PERMISSIONS:", func() {
client := createAuthorizedConnection()
userId := successfullyCreateUser(createRandomString(16), client)
defer deleteUserAfterTests(userId, client)
- request := iggcon.UpdateUserPermissionsRequest{
- UserID: iggcon.NewIdentifier(int(userId)),
- Permissions: &iggcon.Permissions{
+
+ err := client.UpdatePermissions(
+ iggcon.NewIdentifier(int(userId)),
+ &iggcon.Permissions{
Global: iggcon.GlobalPermissions{
ManageServers: false,
ReadServers: false,
@@ -43,10 +44,7 @@ var _ = Describe("UPDATE USER PERMISSIONS:", func() {
PollMessages: false,
SendMessages: false,
},
- },
- }
-
- err := client.UpdateUserPermissions(request)
+ })
itShouldNotReturnError(err)
itShouldSuccessfullyUpdateUserPermissions(userId, client)
@@ -55,13 +53,13 @@ var _ = Describe("UPDATE USER PERMISSIONS:", func() {
When("User is not logged in", func() {
Context("and tries to change user permissions", func() {
- client := createConnection()
- request := iggcon.UpdateUserRequest{
- UserID: iggcon.NewIdentifier(int(createRandomUInt32())),
- Username: createRandomString(16),
- }
-
- err := client.UpdateUser(request)
+ client := createClient()
+ username := createRandomString(16)
+ err := client.UpdateUser(
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ &username,
+ nil,
+ )
itShouldReturnUnauthenticatedError(err)
})
})
diff --git a/bdd/go/tests/tcp_test/users_feature_update.go b/bdd/go/tests/tcp_test/users_feature_update.go
index aa7d6327..de5b3398 100644
--- a/bdd/go/tests/tcp_test/users_feature_update.go
+++ b/bdd/go/tests/tcp_test/users_feature_update.go
@@ -28,27 +28,29 @@ var _ = Describe("UPDATE USER:", func() {
client := createAuthorizedConnection()
userId := successfullyCreateUser(createRandomString(16), client)
defer deleteUserAfterTests(userId, client)
- request := iggcon.UpdateUserRequest{
- UserID: iggcon.NewIdentifier(int(userId)),
- Username: createRandomString(16),
- }
- err := client.UpdateUser(request)
+ username := createRandomString(16)
+ err := client.UpdateUser(
+ iggcon.NewIdentifier(int(userId)),
+ &username,
+ nil,
+ )
itShouldNotReturnError(err)
- itShouldSuccessfullyUpdateUser(userId, request.Username, client)
+ itShouldSuccessfullyUpdateUser(userId, username, client)
})
})
When("User is not logged in", func() {
Context("and tries to update user", func() {
- client := createConnection()
- request := iggcon.UpdateUserRequest{
- UserID: iggcon.NewIdentifier(int(createRandomUInt32())),
- Username: createRandomString(16),
- }
+ client := createClient()
- err := client.UpdateUser(request)
+ username := createRandomString(16)
+ err := client.UpdateUser(
+ iggcon.NewIdentifier(int(createRandomUInt32())),
+ &username,
+ nil,
+ )
itShouldReturnUnauthenticatedError(err)
})
})
diff --git a/bdd/go/tests/tcp_test/users_steps.go b/bdd/go/tests/tcp_test/users_steps.go
index 1d85d7cc..2e26b720 100644
--- a/bdd/go/tests/tcp_test/users_steps.go
+++ b/bdd/go/tests/tcp_test/users_steps.go
@@ -18,20 +18,20 @@
package tcp_test
import (
- "github.com/apache/iggy/foreign/go"
iggcon "github.com/apache/iggy/foreign/go/contracts"
+ "github.com/apache/iggy/foreign/go/iggycli"
. "github.com/onsi/ginkgo/v2"
. "github.com/onsi/gomega"
)
// OPERATIONS
-func successfullyCreateUser(name string, client iggy.MessageStream) uint32 {
- request := iggcon.CreateUserRequest{
- Username: name,
- Password: createRandomString(16),
- Status: iggcon.Active,
- Permissions: &iggcon.Permissions{
+func successfullyCreateUser(name string, client iggycli.Client) uint32 {
+ _, err := client.CreateUser(
+ name,
+ createRandomString(16),
+ iggcon.Active,
+ &iggcon.Permissions{
Global: iggcon.GlobalPermissions{
ManageServers: true,
ReadServers: true,
@@ -44,10 +44,7 @@ func successfullyCreateUser(name string, client iggy.MessageStream) uint32 {
PollMessages: true,
SendMessages: true,
},
- },
- }
-
- err := client.CreateUser(request)
+ })
itShouldNotReturnError(err)
user, err := client.GetUser(iggcon.NewIdentifier(name))
itShouldNotReturnError(err)
@@ -57,7 +54,7 @@ func successfullyCreateUser(name string, client iggy.MessageStream) uint32 {
// ASSERTIONS
-func itShouldSuccessfullyCreateUser(name string, client iggy.MessageStream) {
+func itShouldSuccessfullyCreateUser(name string, client iggycli.Client) {
user, err := client.GetUser(iggcon.NewIdentifier(name))
itShouldNotReturnError(err)
@@ -67,7 +64,7 @@ func itShouldSuccessfullyCreateUser(name string, client iggy.MessageStream) {
})
}
-func itShouldSuccessfullyCreateUserWithPermissions(name string, client iggy.MessageStream, permissions map[int]*iggcon.StreamPermissions) {
+func itShouldSuccessfullyCreateUserWithPermissions(name string, client iggycli.Client, permissions map[int]*iggcon.StreamPermissions) {
user, err := client.GetUser(iggcon.NewIdentifier(name))
itShouldNotReturnError(err)
@@ -97,7 +94,7 @@ func itShouldSuccessfullyCreateUserWithPermissions(name string, client iggy.Mess
})
}
-func itShouldSuccessfullyUpdateUser(id uint32, name string, client iggy.MessageStream) {
+func itShouldSuccessfullyUpdateUser(id uint32, name string, client iggycli.Client) {
user, err := client.GetUser(iggcon.NewIdentifier(name))
itShouldNotReturnError(err)
@@ -111,7 +108,7 @@ func itShouldSuccessfullyUpdateUser(id uint32, name string, client iggy.MessageS
})
}
-func itShouldSuccessfullyDeleteUser(userId int, client iggy.MessageStream) {
+func itShouldSuccessfullyDeleteUser(userId int, client iggycli.Client) {
user, err := client.GetUser(iggcon.NewIdentifier(userId))
itShouldReturnSpecificError(err, "resource_not_found")
@@ -120,7 +117,7 @@ func itShouldSuccessfullyDeleteUser(userId int, client iggy.MessageStream) {
})
}
-func itShouldSuccessfullyUpdateUserPermissions(userId uint32, client iggy.MessageStream) {
+func itShouldSuccessfullyUpdateUserPermissions(userId uint32, client iggycli.Client) {
user, err := client.GetUser(iggcon.NewIdentifier(int(userId)))
itShouldNotReturnError(err)
@@ -140,11 +137,8 @@ func itShouldSuccessfullyUpdateUserPermissions(userId uint32, client iggy.Messag
}
func itShouldBePossibleToLogInWithCredentials(username string, password string) {
- ms := createConnection()
- userId, err := ms.LogIn(iggcon.LogInRequest{
- Username: username,
- Password: password,
- })
+ ms := createClient()
+ userId, err := ms.LoginUser(username, password)
itShouldNotReturnError(err)
It("should return userId", func() {
@@ -152,23 +146,23 @@ func itShouldBePossibleToLogInWithCredentials(username string, password string)
})
}
-func itShouldReturnSpecificUser(name string, user iggcon.UserResponse) {
+func itShouldReturnSpecificUser(name string, user iggcon.UserInfo) {
It("should fetch user with name "+name, func() {
Expect(user.Username).To(Equal(name))
})
}
-func itShouldContainSpecificUser(name string, users []*iggcon.UserResponse) {
+func itShouldContainSpecificUser(name string, users []iggcon.UserInfo) {
It("should fetch at least one user", func() {
Expect(len(users)).NotTo(Equal(0))
})
- var user iggcon.UserResponse
+ var user iggcon.UserInfo
found := false
for _, s := range users {
if s.Username == name {
- user = *s
+ user = s
found = true
break
}
@@ -182,6 +176,6 @@ func itShouldContainSpecificUser(name string, users []*iggcon.UserResponse) {
//CLEANUP
-func deleteUserAfterTests(name any, client iggy.MessageStream) {
+func deleteUserAfterTests(name any, client iggycli.Client) {
_ = client.DeleteUser(iggcon.NewIdentifier(name))
}
diff --git a/foreign/go/benchmarks/send_messages_benchmark_test.go b/foreign/go/benchmarks/send_messages_benchmark_test.go
index 217bf56b..78978526 100644
--- a/foreign/go/benchmarks/send_messages_benchmark_test.go
+++ b/foreign/go/benchmarks/send_messages_benchmark_test.go
@@ -24,9 +24,10 @@ import (
"testing"
"time"
- "github.com/google/uuid"
- "github.com/apache/iggy/foreign/go"
iggcon "github.com/apache/iggy/foreign/go/contracts"
+ "github.com/apache/iggy/foreign/go/iggycli"
+ "github.com/apache/iggy/foreign/go/tcp"
+ "github.com/google/uuid"
)
const (
@@ -40,30 +41,25 @@ const (
func BenchmarkSendMessage(b *testing.B) {
rand.New(rand.NewSource(42)) // Seed the random number generator for consistent results
- streams := make([]iggy.MessageStream, producerCount)
-
- factory := &iggy.IggyClientFactory{}
- config := iggcon.IggyConfiguration{
- BaseAddress: "127.0.0.1:8090",
- Protocol: iggcon.Tcp,
- }
+ clients := make([]iggycli.Client, producerCount)
for i := 0; i < producerCount; i++ {
- ms, err := factory.CreateMessageStream(config)
+ cli, err := iggycli.NewIggyClient(
+ iggycli.WithTcp(
+ tcp.WithServerAddress("127.0.0.1:8090"),
+ ),
+ )
if err != nil {
panic("COULD NOT CREATE MESSAGE STREAM")
}
- _, err = ms.LogIn(iggcon.LogInRequest{
- Username: "iggy",
- Password: "iggy",
- })
+ _, err = cli.LoginUser("iggy", "iggy")
if err != nil {
panic("COULD NOT LOG IN")
}
- streams[i] = ms
+ clients[i] = cli
}
- for index, value := range streams {
+ for index, value := range clients {
err := ensureInfrastructureIsInitialized(value, startingStreamId+index)
if err != nil {
panic("COULD NOT INITIALIZE INFRASTRUCTURE")
@@ -80,7 +76,7 @@ func BenchmarkSendMessage(b *testing.B) {
wg.Add(1)
go func(i int) {
defer wg.Done()
- avgLatency, avgThroughput := SendMessage(streams[i], i, messagesCount, messagesBatch, messageSize)
+ avgLatency, avgThroughput := SendMessage(clients[i], i, messagesCount, messagesBatch, messageSize)
resultChannel <- struct {
avgLatency float64
@@ -99,7 +95,7 @@ func BenchmarkSendMessage(b *testing.B) {
// Print the final results
fmt.Printf("Summarized Average Throughput: %.2f MB/s\n", aggregateThroughput)
- for index, value := range streams {
+ for index, value := range clients {
err := cleanupInfrastructure(value, startingStreamId+index)
if err != nil {
panic("COULD NOT CLEANUP INFRASTRUCTURE")
@@ -107,25 +103,25 @@ func BenchmarkSendMessage(b *testing.B) {
}
}
-func ensureInfrastructureIsInitialized(messageStream iggy.MessageStream, streamId int) error {
- if _, streamErr := messageStream.GetStreamById(iggcon.GetStreamRequest{
- StreamID: iggcon.NewIdentifier(streamId)}); streamErr != nil {
- streamErr = messageStream.CreateStream(iggcon.CreateStreamRequest{
- StreamId: streamId,
- Name: "benchmark" + fmt.Sprint(streamId),
- })
+func ensureInfrastructureIsInitialized(cli iggycli.Client, streamId int) error {
+ if _, streamErr := cli.GetStream(iggcon.NewIdentifier(streamId)); streamErr != nil {
+ uint32StreamId := uint32(streamId)
+ _, streamErr = cli.CreateStream("benchmark"+fmt.Sprint(streamId), &uint32StreamId)
if streamErr != nil {
panic(streamErr)
}
}
- if _, topicErr := messageStream.GetTopicById(iggcon.NewIdentifier(streamId), iggcon.NewIdentifier(1)); topicErr != nil {
- topicErr = messageStream.CreateTopic(iggcon.CreateTopicRequest{
- TopicId: 1,
- Name: "benchmark",
- PartitionsCount: 1,
- StreamId: iggcon.NewIdentifier(streamId),
- CompressionAlgorithm: 1,
- })
+ if _, topicErr := cli.GetTopic(iggcon.NewIdentifier(streamId), iggcon.NewIdentifier(1)); topicErr != nil {
+ _, topicErr = cli.CreateTopic(
+ iggcon.NewIdentifier(streamId),
+ "benchmark",
+ 1,
+ 1,
+ 0,
+ 1,
+ nil,
+ nil,
+ )
if topicErr != nil {
panic(topicErr)
@@ -134,8 +130,8 @@ func ensureInfrastructureIsInitialized(messageStream iggy.MessageStream, streamI
return nil
}
-func cleanupInfrastructure(messageStream iggy.MessageStream, streamId int) error {
- return messageStream.DeleteStream(iggcon.NewIdentifier(streamId))
+func cleanupInfrastructure(cli iggycli.Client, streamId int) error {
+ return cli.DeleteStream(iggcon.NewIdentifier(streamId))
}
// CreateMessages creates messages with random payloads.
@@ -158,24 +154,23 @@ func CreateMessages(messagesCount, messageSize int) []iggcon.IggyMessage {
}
// SendMessage performs the message sending operation.
-func SendMessage(bus iggy.MessageStream, producerNumber, messagesCount, messagesBatch, messageSize int) (avgLatency float64, avgThroughput float64) {
+func SendMessage(cli iggycli.Client, producerNumber, messagesCount, messagesBatch, messageSize int) (avgLatency float64, avgThroughput float64) {
totalMessages := messagesBatch * messagesCount
totalMessagesBytes := int64(totalMessages * messageSize)
fmt.Printf("Executing Send Messages command for producer %d, messages count %d, with size %d bytes\n", producerNumber, totalMessages, totalMessagesBytes)
- busStreamId := iggcon.NewIdentifier(startingStreamId + producerNumber)
- busTopicId := iggcon.NewIdentifier(topicId)
+ streamId := iggcon.NewIdentifier(startingStreamId + producerNumber)
messages := CreateMessages(messagesCount, messageSize)
latencies := make([]time.Duration, 0)
for i := 0; i < messagesBatch; i++ {
startTime := time.Now()
- _ = bus.SendMessages(iggcon.SendMessagesRequest{
- StreamId: busStreamId,
- TopicId: busTopicId,
- Partitioning: iggcon.PartitionId(1),
- Messages: messages,
- })
+ _ = cli.SendMessages(
+ streamId,
+ iggcon.NewIdentifier(topicId),
+ iggcon.PartitionId(1),
+ messages,
+ )
elapsedTime := time.Since(startTime)
latencies = append(latencies, elapsedTime)
}
diff --git a/foreign/go/binary_serialization/binary_request_serializer.go b/foreign/go/binary_serialization/binary_request_serializer.go
index 4cdd9220..d1b7e9c6 100644
--- a/foreign/go/binary_serialization/binary_request_serializer.go
+++ b/foreign/go/binary_serialization/binary_request_serializer.go
@@ -24,32 +24,41 @@ import (
)
func CreateGroup(request CreateConsumerGroupRequest) []byte {
+ if request.ConsumerGroupId == nil {
+ request.ConsumerGroupId = new(uint32)
+ }
customIdOffset := 4 + request.StreamId.Length + request.TopicId.Length
bytes := make([]byte, 4+request.StreamId.Length+request.TopicId.Length+1+4+len(request.Name))
copy(bytes[0:customIdOffset], SerializeIdentifiers(request.StreamId, request.TopicId))
- binary.LittleEndian.PutUint32(bytes[customIdOffset:customIdOffset+4], uint32(request.ConsumerGroupId))
+ binary.LittleEndian.PutUint32(bytes[customIdOffset:customIdOffset+4], *request.ConsumerGroupId)
bytes[customIdOffset+4] = byte(len(request.Name))
- copy(bytes[customIdOffset+5:], []byte(request.Name))
+ copy(bytes[customIdOffset+5:], request.Name)
return bytes
}
-func UpdateOffset(request StoreOffsetRequest) []byte {
+func UpdateOffset(request StoreConsumerOffsetRequest) []byte {
+ if request.PartitionId == nil {
+ request.PartitionId = new(uint32)
+ }
bytes := make([]byte, 6+request.StreamId.Length+request.TopicId.Length+request.Consumer.Id.Length+13)
bytes[0] = byte(request.Consumer.Kind)
position := 7 + request.StreamId.Length + request.TopicId.Length + request.Consumer.Id.Length
copy(bytes[1:position], SerializeIdentifiers(request.Consumer.Id, request.StreamId, request.TopicId))
- binary.LittleEndian.PutUint32(bytes[position:position+4], uint32(request.PartitionId))
+ binary.LittleEndian.PutUint32(bytes[position:position+4], *request.PartitionId)
binary.LittleEndian.PutUint64(bytes[position+4:position+12], uint64(request.Offset))
return bytes
}
-func GetOffset(request GetOffsetRequest) []byte {
+func GetOffset(request GetConsumerOffsetRequest) []byte {
+ if request.PartitionId == nil {
+ request.PartitionId = new(uint32)
+ }
bytes := make([]byte, 6+request.StreamId.Length+request.TopicId.Length+request.Consumer.Id.Length+5)
bytes[0] = byte(request.Consumer.Kind)
position := 7 + request.StreamId.Length + request.TopicId.Length + request.Consumer.Id.Length
copy(bytes[1:position], SerializeIdentifiers(request.Consumer.Id, request.StreamId, request.TopicId))
- binary.LittleEndian.PutUint32(bytes[position:position+4], uint32(request.PartitionId))
+ binary.LittleEndian.PutUint32(bytes[position:position+4], *request.PartitionId)
return bytes
}
@@ -62,7 +71,7 @@ func CreatePartitions(request CreatePartitionsRequest) []byte {
return bytes
}
-func DeletePartitions(request DeletePartitionRequest) []byte {
+func DeletePartitions(request DeletePartitionsRequest) []byte {
bytes := make([]byte, 8+request.StreamId.Length+request.TopicId.Length)
position := 4 + request.StreamId.Length + request.TopicId.Length
copy(bytes[0:position], SerializeIdentifiers(request.StreamId, request.TopicId))
@@ -214,8 +223,14 @@ func boolToByte(b bool) byte {
func SerializeUpdateUser(request UpdateUserRequest) []byte {
length := request.UserID.Length + 2
- if len(request.Username) != 0 {
- length += 2 + len(request.Username)
+ if request.Username == nil {
+ request.Username = new(string)
+ }
+
+ username := *request.Username
+
+ if len(username) != 0 {
+ length += 2 + len(username)
}
if request.Status != nil {
@@ -228,13 +243,13 @@ func SerializeUpdateUser(request UpdateUserRequest) []byte {
copy(bytes[position:position+request.UserID.Length+2], SerializeIdentifier(request.UserID))
position += position + request.UserID.Length + 2
- if len(request.Username) != 0 {
+ if len(username) != 0 {
bytes[position] = 1
position++
- bytes[position] = byte(len(request.Username))
+ bytes[position] = byte(len(username))
position++
- copy(bytes[position:position+len(request.Username)], []byte(request.Username))
- position += len(request.Username)
+ copy(bytes[position:position+len(username)], username)
+ position += len(username)
} else {
bytes[position] = 0
position++
@@ -278,7 +293,7 @@ func SerializeChangePasswordRequest(request ChangePasswordRequest) []byte {
return bytes
}
-func SerializeUpdateUserPermissionsRequest(request UpdateUserPermissionsRequest) []byte {
+func SerializeUpdateUserPermissionsRequest(request UpdatePermissionsRequest) []byte {
length := request.UserID.Length + 2
if request.Permissions != nil {
@@ -313,7 +328,7 @@ func SerializeInt(value int) []byte {
return bytes
}
-func SerializeLoginWithPersonalAccessToken(request LogInAccessTokenRequest) []byte {
+func SerializeLoginWithPersonalAccessToken(request LoginWithPersonalAccessTokenRequest) []byte {
length := 1 + len(request.Token)
bytes := make([]byte, length)
bytes[0] = byte(len(request.Token))
@@ -321,7 +336,7 @@ func SerializeLoginWithPersonalAccessToken(request LogInAccessTokenRequest) []by
return bytes
}
-func SerializeDeletePersonalAccessToken(request DeleteAccessTokenRequest) []byte {
+func SerializeDeletePersonalAccessToken(request DeletePersonalAccessTokenRequest) []byte {
length := 1 + len(request.Name)
bytes := make([]byte, length)
bytes[0] = byte(len(request.Name))
@@ -329,7 +344,7 @@ func SerializeDeletePersonalAccessToken(request DeleteAccessTokenRequest) []byte
return bytes
}
-func SerializeCreatePersonalAccessToken(request CreateAccessTokenRequest) []byte {
+func SerializeCreatePersonalAccessToken(request CreatePersonalAccessTokenRequest) []byte {
length := 1 + len(request.Name) + 8
bytes := make([]byte, length)
bytes[0] = byte(len(request.Name))
diff --git a/foreign/go/binary_serialization/binary_response_deserializer.go b/foreign/go/binary_serialization/binary_response_deserializer.go
index 1c1d7abf..8c1fd51d 100644
--- a/foreign/go/binary_serialization/binary_response_deserializer.go
+++ b/foreign/go/binary_serialization/binary_response_deserializer.go
@@ -28,27 +28,36 @@ import (
"github.com/klauspost/compress/s2"
)
-func DeserializeLogInResponse(payload []byte) *LogInResponse {
+func DeserializeLogInResponse(payload []byte) *IdentityInfo {
userId := binary.LittleEndian.Uint32(payload[0:4])
- return &LogInResponse{
+ return &IdentityInfo{
UserId: userId,
}
}
-func DeserializeOffset(payload []byte) *OffsetResponse {
+func DeserializeOffset(payload []byte) *ConsumerOffsetInfo {
partitionId := int(binary.LittleEndian.Uint32(payload[0:4]))
currentOffset := binary.LittleEndian.Uint64(payload[4:12])
storedOffset := binary.LittleEndian.Uint64(payload[12:20])
- return &OffsetResponse{
+ return &ConsumerOffsetInfo{
PartitionId: partitionId,
CurrentOffset: currentOffset,
StoredOffset: storedOffset,
}
}
-func DeserializeStreams(payload []byte) []StreamResponse {
- streams := make([]StreamResponse, 0)
+func DeserializeStream(payload []byte) *StreamDetails {
+ stream, _ := DeserializeToStream(payload, 0)
+ // TODO implement deserialize topics
+ return &StreamDetails{
+ Stream: stream,
+ Topics: nil,
+ }
+}
+
+func DeserializeStreams(payload []byte) []Stream {
+ streams := make([]Stream, 0)
position := 0
//TODO there's a deserialization bug, investigate this
@@ -62,29 +71,7 @@ func DeserializeStreams(payload []byte) []StreamResponse {
return streams
}
-func DeserializerStream(payload []byte) *StreamResponse {
- stream, position := DeserializeToStream(payload, 0)
- topics := make([]TopicResponse, 0)
- length := len(payload)
-
- for position < length {
- topic, readBytes, _ := DeserializeToTopic(payload, position)
- topics = append(topics, topic)
- position += readBytes
- }
-
- return &StreamResponse{
- Id: stream.Id,
- TopicsCount: stream.TopicsCount,
- Name: stream.Name,
- Topics: topics,
- MessagesCount: stream.MessagesCount,
- SizeBytes: stream.SizeBytes,
- CreatedAt: stream.CreatedAt,
- }
-}
-
-func DeserializeToStream(payload []byte, position int) (StreamResponse, int) {
+func DeserializeToStream(payload []byte, position int) (Stream, int) {
id := int(binary.LittleEndian.Uint32(payload[position : position+4]))
createdAt := binary.LittleEndian.Uint64(payload[position+4 : position+12])
topicsCount := int(binary.LittleEndian.Uint32(payload[position+12 : position+16]))
@@ -97,7 +84,7 @@ func DeserializeToStream(payload []byte, position int) (StreamResponse, int) {
readBytes := 4 + 8 + 4 + 8 + 8 + 1 + nameLength
- return StreamResponse{
+ return Stream{
Id: id,
TopicsCount: topicsCount,
Name: name,
@@ -107,9 +94,9 @@ func DeserializeToStream(payload []byte, position int) (StreamResponse, int) {
}, readBytes
}
-func DeserializeFetchMessagesResponse(payload []byte, compression IggyMessageCompression) (*FetchMessagesResponse, error) {
+func DeserializeFetchMessagesResponse(payload []byte, compression IggyMessageCompression) (*PolledMessage, error) {
if len(payload) == 0 {
- return &FetchMessagesResponse{
+ return &PolledMessage{
PartitionId: 0,
CurrentOffset: 0,
Messages: make([]IggyMessage, 0),
@@ -164,7 +151,7 @@ func DeserializeFetchMessagesResponse(payload []byte, compression IggyMessageCom
}
// !TODO: Add message offset ordering
- return &FetchMessagesResponse{
+ return &PolledMessage{
PartitionId: partitionId,
CurrentOffset: currentOffset,
Messages: messages,
@@ -172,8 +159,8 @@ func DeserializeFetchMessagesResponse(payload []byte, compression IggyMessageCom
}, nil
}
-func DeserializeTopics(payload []byte) ([]TopicResponse, error) {
- topics := make([]TopicResponse, 0)
+func DeserializeTopics(payload []byte) ([]Topic, error) {
+ topics := make([]Topic, 0)
length := len(payload)
position := 0
@@ -189,10 +176,10 @@ func DeserializeTopics(payload []byte) ([]TopicResponse, error) {
return topics, nil
}
-func DeserializeTopic(payload []byte) (*TopicResponse, error) {
+func DeserializeTopic(payload []byte) (*TopicDetails, error) {
topic, position, err := DeserializeToTopic(payload, 0)
if err != nil {
- return &TopicResponse{}, err
+ return &TopicDetails{}, err
}
partitions := make([]PartitionContract, 0)
@@ -200,20 +187,17 @@ func DeserializeTopic(payload []byte) (*TopicResponse, error) {
for position < length {
partition, readBytes := DeserializePartition(payload, position)
- if err != nil {
- return &TopicResponse{}, err
- }
partitions = append(partitions, partition)
position += readBytes
}
-
- topic.Partitions = partitions
-
- return &topic, nil
+ return &TopicDetails{
+ Topic: topic,
+ Partitions: partitions,
+ }, nil
}
-func DeserializeToTopic(payload []byte, position int) (TopicResponse, int, error) {
- topic := TopicResponse{}
+func DeserializeToTopic(payload []byte, position int) (Topic, int, error) {
+ topic := Topic{}
topic.Id = int(binary.LittleEndian.Uint32(payload[position : position+4]))
topic.CreatedAt = int(binary.LittleEndian.Uint64(payload[position+4 : position+12]))
topic.PartitionsCount = int(binary.LittleEndian.Uint32(payload[position+12 : position+16]))
@@ -252,8 +236,8 @@ func DeserializePartition(payload []byte, position int) (PartitionContract, int)
return partition, readBytes
}
-func DeserializeConsumerGroups(payload []byte) []ConsumerGroupResponse {
- var consumerGroups []ConsumerGroupResponse
+func DeserializeConsumerGroups(payload []byte) []ConsumerGroup {
+ var consumerGroups []ConsumerGroup
length := len(payload)
position := 0
@@ -267,12 +251,7 @@ func DeserializeConsumerGroups(payload []byte) []ConsumerGroupResponse {
return consumerGroups
}
-func DeserializeConsumerGroup(payload []byte) (*ConsumerGroupResponse, error) {
- consumerGroup, _ := DeserializeToConsumerGroup(payload, 0)
- return consumerGroup, nil
-}
-
-func DeserializeToConsumerGroup(payload []byte, position int) (*ConsumerGroupResponse, int) {
+func DeserializeToConsumerGroup(payload []byte, position int) (*ConsumerGroup, int) {
id := int(binary.LittleEndian.Uint32(payload[position : position+4]))
partitionsCount := int(binary.LittleEndian.Uint32(payload[position+4 : position+8]))
membersCount := int(binary.LittleEndian.Uint32(payload[position+8 : position+12]))
@@ -281,7 +260,7 @@ func DeserializeToConsumerGroup(payload []byte, position int) (*ConsumerGroupRes
readBytes := 12 + 1 + nameLength
- consumerGroup := ConsumerGroupResponse{
+ consumerGroup := ConsumerGroup{
Id: id,
MembersCount: membersCount,
PartitionsCount: partitionsCount,
@@ -291,47 +270,59 @@ func DeserializeToConsumerGroup(payload []byte, position int) (*ConsumerGroupRes
return &consumerGroup, readBytes
}
-func DeserializeUsers(payload []byte) ([]*UserResponse, error) {
+func DeserializeConsumerGroup(payload []byte) *ConsumerGroupDetails {
+ consumerGroup, _ := DeserializeToConsumerGroup(payload, 0)
+ // TODO: implement logic to deserialize the members.
+ return &ConsumerGroupDetails{
+ ConsumerGroup: *consumerGroup,
+ Members: nil,
+ }
+}
+
+func DeserializeUsers(payload []byte) ([]UserInfo, error) {
if len(payload) == 0 {
return nil, errors.New("empty payload")
}
- var result []*UserResponse
+ var result []UserInfo
length := len(payload)
position := 0
for position < length {
- response, readBytes, err := deserializeUserResponse(payload, position)
+ response, readBytes, err := deserializeToUser(payload, position)
if err != nil {
return nil, err
}
- result = append(result, response)
+ result = append(result, *response)
position += readBytes
}
return result, nil
}
-func DeserializeUser(payload []byte) (*UserResponse, error) {
- response, position, err := deserializeUserResponse(payload, 0)
+func DeserializeUser(payload []byte) (*UserInfoDetails, error) {
+ response, position, err := deserializeToUser(payload, 0)
+ if err != nil {
+ return nil, err
+ }
hasPermissions := payload[position]
+ userInfo := UserInfo{
+ Id: response.Id,
+ CreatedAt: response.CreatedAt,
+ Username: response.Username,
+ Status: response.Status,
+ }
if hasPermissions == 1 {
permissionLength := binary.LittleEndian.Uint32(payload[position+1 : position+5])
permissionsPayload := payload[position+5 : position+5+int(permissionLength)]
permissions := deserializePermissions(permissionsPayload)
- return &UserResponse{
+ return &UserInfoDetails{
+ UserInfo: userInfo,
Permissions: permissions,
- Id: response.Id,
- CreatedAt: response.CreatedAt,
- Username: response.Username,
- Status: response.Status,
}, err
}
- return &UserResponse{
- Id: response.Id,
- CreatedAt: response.CreatedAt,
- Username: response.Username,
- Status: response.Status,
+ return &UserInfoDetails{
+ UserInfo: userInfo,
Permissions: nil,
}, err
}
@@ -421,9 +412,9 @@ func deserializePermissions(bytes []byte) *Permissions {
}
}
-func deserializeUserResponse(payload []byte, position int) (*UserResponse, int, error) {
+func deserializeToUser(payload []byte, position int) (*UserInfo, int, error) {
if len(payload) < position+14 {
- return nil, 0, errors.New("not enough data to map UserResponse")
+ return nil, 0, errors.New("not enough data to map UserInfo")
}
id := binary.LittleEndian.Uint32(payload[position : position+4])
@@ -447,7 +438,7 @@ func deserializeUserResponse(payload []byte, position int) (*UserResponse, int,
readBytes := 4 + 8 + 1 + 1 + int(usernameLength)
- return &UserResponse{
+ return &UserInfo{
Id: id,
CreatedAt: createdAt,
Status: userStatus,
@@ -455,12 +446,12 @@ func deserializeUserResponse(payload []byte, position int) (*UserResponse, int,
}, readBytes, nil
}
-func DeserializeClients(payload []byte) ([]ClientResponse, error) {
+func DeserializeClients(payload []byte) ([]ClientInfo, error) {
if len(payload) == 0 {
- return []ClientResponse{}, nil
+ return []ClientInfo{}, nil
}
- var response []ClientResponse
+ var response []ClientInfo
length := len(payload)
position := 0
@@ -473,7 +464,7 @@ func DeserializeClients(payload []byte) ([]ClientResponse, error) {
return response, nil
}
-func MapClientInfo(payload []byte, position int) (ClientResponse, int) {
+func MapClientInfo(payload []byte, position int) (ClientInfo, int) {
var readBytes int
id := binary.LittleEndian.Uint32(payload[position : position+4])
userId := binary.LittleEndian.Uint32(payload[position+4 : position+8])
@@ -493,7 +484,7 @@ func MapClientInfo(payload []byte, position int) (ClientResponse, int) {
consumerGroupsCount := binary.LittleEndian.Uint32(payload[position : position+4])
readBytes += 4
- return ClientResponse{
+ return ClientInfo{
ID: id,
UserID: userId,
Transport: transport,
@@ -502,13 +493,13 @@ func MapClientInfo(payload []byte, position int) (ClientResponse, int) {
}, readBytes
}
-func DeserializeClient(payload []byte) *ClientResponse {
- response, position := MapClientInfo(payload, 0)
- consumerGroups := make([]ConsumerGroupInfo, response.ConsumerGroupsCount)
+func DeserializeClient(payload []byte) *ClientInfoDetails {
+ clientInfo, position := MapClientInfo(payload, 0)
+ consumerGroups := make([]ConsumerGroupInfo, clientInfo.ConsumerGroupsCount)
length := len(payload)
for position < length {
- for i := uint32(0); i < response.ConsumerGroupsCount; i++ {
+ for i := uint32(0); i < clientInfo.ConsumerGroupsCount; i++ {
streamId := int32(binary.LittleEndian.Uint32(payload[position : position+4]))
topicId := int32(binary.LittleEndian.Uint32(payload[position+4 : position+8]))
consumerGroupId := int32(binary.LittleEndian.Uint32(payload[position+8 : position+12]))
@@ -522,24 +513,26 @@ func DeserializeClient(payload []byte) *ClientResponse {
position += 12
}
}
- response.ConsumerGroups = consumerGroups
- return &response
+ return &ClientInfoDetails{
+ ClientInfo: clientInfo,
+ ConsumerGroups: consumerGroups,
+ }
}
-func DeserializeAccessToken(payload []byte) (*AccessToken, error) {
+func DeserializeAccessToken(payload []byte) (*RawPersonalAccessToken, error) {
tokenLength := int(payload[0])
token := string(payload[1 : 1+tokenLength])
- return &AccessToken{
+ return &RawPersonalAccessToken{
Token: token,
}, nil
}
-func DeserializeAccessTokens(payload []byte) ([]AccessTokenResponse, error) {
+func DeserializeAccessTokens(payload []byte) ([]PersonalAccessTokenInfo, error) {
if len(payload) == 0 {
- return []AccessTokenResponse{}, ierror.CustomError("Empty payload")
+ return []PersonalAccessTokenInfo{}, ierror.CustomError("Empty payload")
}
- var result []AccessTokenResponse
+ var result []PersonalAccessTokenInfo
position := 0
length := len(payload)
@@ -552,7 +545,7 @@ func DeserializeAccessTokens(payload []byte) ([]AccessTokenResponse, error) {
return result, nil
}
-func deserializeToPersonalAccessTokenResponse(payload []byte, position int) (AccessTokenResponse, int) {
+func deserializeToPersonalAccessTokenResponse(payload []byte, position int) (PersonalAccessTokenInfo, int) {
nameLength := int(payload[position])
name := string(payload[position+1 : position+1+nameLength])
expiryBytes := payload[position+1+nameLength:]
@@ -566,7 +559,7 @@ func deserializeToPersonalAccessTokenResponse(payload []byte, position int) (Acc
readBytes := 1 + nameLength + 8
- return AccessTokenResponse{
+ return PersonalAccessTokenInfo{
Name: name,
Expiry: expiry,
}, readBytes
diff --git a/foreign/go/binary_serialization/create_stream_serializer.go b/foreign/go/binary_serialization/create_stream_serializer.go
index 002d669f..a9deb232 100644
--- a/foreign/go/binary_serialization/create_stream_serializer.go
+++ b/foreign/go/binary_serialization/create_stream_serializer.go
@@ -19,12 +19,11 @@ package binaryserialization
import (
"encoding/binary"
-
- iggcon "github.com/apache/iggy/foreign/go/contracts"
)
type TcpCreateStreamRequest struct {
- iggcon.CreateStreamRequest
+ Name string
+ StreamId *uint32
}
const (
@@ -34,10 +33,14 @@ const (
)
func (request *TcpCreateStreamRequest) Serialize() []byte {
+ if request.StreamId == nil {
+ request.StreamId = new(uint32)
+ }
+
nameLength := len(request.Name)
serialized := make([]byte, payloadOffset+nameLength)
- binary.LittleEndian.PutUint32(serialized[streamIDOffset:], uint32(request.StreamId))
+ binary.LittleEndian.PutUint32(serialized[streamIDOffset:], *request.StreamId)
serialized[nameLengthOffset] = byte(nameLength)
copy(serialized[payloadOffset:], []byte(request.Name))
diff --git a/foreign/go/binary_serialization/create_stream_serializer_test.go b/foreign/go/binary_serialization/create_stream_serializer_test.go
index 81d85db3..65f1f788 100644
--- a/foreign/go/binary_serialization/create_stream_serializer_test.go
+++ b/foreign/go/binary_serialization/create_stream_serializer_test.go
@@ -19,18 +19,16 @@ package binaryserialization
import (
"encoding/binary"
- iggcon "github.com/apache/iggy/foreign/go/contracts"
"reflect"
"testing"
)
func TestSerialize_TcpCreateStreamRequest(t *testing.T) {
// Create a sample TcpCreateStreamRequest
+ streamId := uint32(123)
request := TcpCreateStreamRequest{
- CreateStreamRequest: iggcon.CreateStreamRequest{
- StreamId: 123,
- Name: "test_stream",
- },
+ StreamId: &streamId,
+ Name: "test_stream",
}
// Serialize the request
diff --git a/foreign/go/binary_serialization/create_topic_serializer.go b/foreign/go/binary_serialization/create_topic_serializer.go
index ef2d101a..c3d21a66 100644
--- a/foreign/go/binary_serialization/create_topic_serializer.go
+++ b/foreign/go/binary_serialization/create_topic_serializer.go
@@ -19,15 +19,30 @@ package binaryserialization
import (
"encoding/binary"
+ "time"
iggcon "github.com/apache/iggy/foreign/go/contracts"
)
type TcpCreateTopicRequest struct {
- iggcon.CreateTopicRequest
+ StreamId iggcon.Identifier `json:"streamId"`
+ PartitionsCount int `json:"partitionsCount"`
+ CompressionAlgorithm uint8 `json:"compressionAlgorithm"`
+ MessageExpiry time.Duration `json:"messageExpiry"`
+ MaxTopicSize uint64 `json:"maxTopicSize"`
+ Name string `json:"name"`
+ ReplicationFactor *uint8 `json:"replicationFactor"`
+ TopicId *int `json:"topicId"`
}
func (request *TcpCreateTopicRequest) Serialize() []byte {
+ if request.TopicId == nil {
+ request.TopicId = new(int)
+ }
+ if request.ReplicationFactor == nil {
+ request.ReplicationFactor = new(uint8)
+ }
+
streamIdBytes := SerializeIdentifier(request.StreamId)
nameBytes := []byte(request.Name)
@@ -49,7 +64,7 @@ func (request *TcpCreateTopicRequest) Serialize() []byte {
position += len(streamIdBytes)
// TopicId
- binary.LittleEndian.PutUint32(bytes[position:], uint32(request.TopicId))
+ binary.LittleEndian.PutUint32(bytes[position:], uint32(*request.TopicId))
position += 4
// PartitionsCount
@@ -69,7 +84,7 @@ func (request *TcpCreateTopicRequest) Serialize() []byte {
position += 8
// ReplicationFactor
- bytes[position] = request.ReplicationFactor
+ bytes[position] = *request.ReplicationFactor
position++
// Name
diff --git a/foreign/go/binary_serialization/fetch_messages_request_serializer.go b/foreign/go/binary_serialization/fetch_messages_request_serializer.go
index 01e0616f..63512eaf 100644
--- a/foreign/go/binary_serialization/fetch_messages_request_serializer.go
+++ b/foreign/go/binary_serialization/fetch_messages_request_serializer.go
@@ -30,10 +30,19 @@ const (
)
type TcpFetchMessagesRequest struct {
- iggcon.FetchMessagesRequest
+ StreamId iggcon.Identifier `json:"streamId"`
+ TopicId iggcon.Identifier `json:"topicId"`
+ Consumer iggcon.Consumer `json:"consumer"`
+ PartitionId *uint32 `json:"partitionId"`
+ Strategy iggcon.PollingStrategy `json:"pollingStrategy"`
+ Count uint32 `json:"count"`
+ AutoCommit bool `json:"autoCommit"`
}
func (request *TcpFetchMessagesRequest) Serialize() []byte {
+ if request.PartitionId == nil {
+ request.PartitionId = new(uint32)
+ }
streamTopicIdLength := 2 + request.StreamId.Length + 2 + request.TopicId.Length
messageSize := 2 + request.Consumer.Id.Length + streamTopicIdLength + partitionStrategySize + offsetSize + commitFlagSize + 1
bytes := make([]byte, messageSize)
@@ -46,12 +55,12 @@ func (request *TcpFetchMessagesRequest) Serialize() []byte {
copy(bytes[position:position+streamTopicIdLength], SerializeIdentifiers(request.StreamId, request.TopicId))
position += streamTopicIdLength
- binary.LittleEndian.PutUint32(bytes[position:position+4], uint32(request.PartitionId))
- bytes[position+4] = byte(request.PollingStrategy.Kind)
+ binary.LittleEndian.PutUint32(bytes[position:position+4], *request.PartitionId)
+ bytes[position+4] = byte(request.Strategy.Kind)
position += partitionStrategySize
- binary.LittleEndian.PutUint64(bytes[position:position+8], uint64(request.PollingStrategy.Value))
- binary.LittleEndian.PutUint32(bytes[position+8:position+12], uint32(request.Count))
+ binary.LittleEndian.PutUint64(bytes[position:position+8], request.Strategy.Value)
+ binary.LittleEndian.PutUint32(bytes[position+8:position+12], request.Count)
position += offsetSize
diff --git a/foreign/go/binary_serialization/fetch_messages_request_serializer_test.go b/foreign/go/binary_serialization/fetch_messages_request_serializer_test.go
index 5ce5b8ce..bf511c78 100644
--- a/foreign/go/binary_serialization/fetch_messages_request_serializer_test.go
+++ b/foreign/go/binary_serialization/fetch_messages_request_serializer_test.go
@@ -24,20 +24,19 @@ import (
)
func TestSerialize_TcpFetchMessagesRequest(t *testing.T) {
+ partitionId := uint32(123)
// Create a sample TcpFetchMessagesRequest
request := TcpFetchMessagesRequest{
- FetchMessagesRequest: iggcon.FetchMessagesRequest{
- Consumer: iggcon.Consumer{
- Kind: iggcon.ConsumerSingle,
- Id: iggcon.NewIdentifier(42),
- },
- StreamId: iggcon.NewIdentifier("test_stream_id"),
- TopicId: iggcon.NewIdentifier("test_topic_id"),
- PartitionId: 123,
- PollingStrategy: iggcon.FirstPollingStrategy(),
- Count: 100,
- AutoCommit: true,
+ Consumer: iggcon.Consumer{
+ Kind: iggcon.ConsumerKindSingle,
+ Id: iggcon.NewIdentifier(42),
},
+ StreamId: iggcon.NewIdentifier("test_stream_id"),
+ TopicId: iggcon.NewIdentifier("test_topic_id"),
+ PartitionId: &partitionId,
+ Strategy: iggcon.FirstPollingStrategy(),
+ Count: 100,
+ AutoCommit: true,
}
// Serialize the request
diff --git a/foreign/go/binary_serialization/log_in_request_serializer.go b/foreign/go/binary_serialization/log_in_request_serializer.go
index 68ed4311..bd714be6 100644
--- a/foreign/go/binary_serialization/log_in_request_serializer.go
+++ b/foreign/go/binary_serialization/log_in_request_serializer.go
@@ -19,19 +19,18 @@ package binaryserialization
import (
"encoding/binary"
-
- iggcon "github.com/apache/iggy/foreign/go/contracts"
)
type TcpLogInRequest struct {
- iggcon.LogInRequest
+ Username string `json:"username"`
+ Password string `json:"password"`
}
func (request *TcpLogInRequest) Serialize() []byte {
usernameBytes := []byte(request.Username)
passwordBytes := []byte(request.Password)
- versionBytes := []byte(request.Version)
- contextBytes := []byte(request.Context)
+ versionBytes := []byte("")
+ contextBytes := []byte("")
// Calculate total length
totalLength := 2 + len(usernameBytes) + len(passwordBytes) +
diff --git a/foreign/go/binary_serialization/send_messages_request_serializer.go b/foreign/go/binary_serialization/send_messages_request_serializer.go
index 9f5aae81..87d47df5 100644
--- a/foreign/go/binary_serialization/send_messages_request_serializer.go
+++ b/foreign/go/binary_serialization/send_messages_request_serializer.go
@@ -25,7 +25,10 @@ import (
)
type TcpSendMessagesRequest struct {
- iggcon.SendMessagesRequest
+ StreamId iggcon.Identifier `json:"streamId"`
+ TopicId iggcon.Identifier `json:"topicId"`
+ Partitioning iggcon.Partitioning `json:"partitioning"`
+ Messages []iggcon.IggyMessage `json:"messages"`
}
const indexSize = 16
diff --git a/foreign/go/binary_serialization/send_messages_request_serializer_test.go b/foreign/go/binary_serialization/send_messages_request_serializer_test.go
index b2d6983c..4bf905f7 100644
--- a/foreign/go/binary_serialization/send_messages_request_serializer_test.go
+++ b/foreign/go/binary_serialization/send_messages_request_serializer_test.go
@@ -20,20 +20,18 @@ package binaryserialization
import (
"testing"
- "github.com/google/uuid"
iggcon "github.com/apache/iggy/foreign/go/contracts"
+ "github.com/google/uuid"
)
func TestSerialize_SendMessagesRequest(t *testing.T) {
message1 := generateTestMessage("data1")
request := TcpSendMessagesRequest{
- SendMessagesRequest: iggcon.SendMessagesRequest{
- StreamId: iggcon.NewIdentifier("test_stream_id"),
- TopicId: iggcon.NewIdentifier("test_topic_id"),
- Partitioning: iggcon.PartitionId(1),
- Messages: []iggcon.IggyMessage{
- message1,
- },
+ StreamId: iggcon.NewIdentifier("test_stream_id"),
+ TopicId: iggcon.NewIdentifier("test_topic_id"),
+ Partitioning: iggcon.PartitionId(1),
+ Messages: []iggcon.IggyMessage{
+ message1,
},
}
diff --git a/foreign/go/binary_serialization/update_stream_serializer.go b/foreign/go/binary_serialization/update_stream_serializer.go
index 83b8fc02..2e179b0c 100644
--- a/foreign/go/binary_serialization/update_stream_serializer.go
+++ b/foreign/go/binary_serialization/update_stream_serializer.go
@@ -22,7 +22,8 @@ import (
)
type TcpUpdateStreamRequest struct {
- iggcon.UpdateStreamRequest
+ StreamId iggcon.Identifier `json:"streamId"`
+ Name string `json:"name"`
}
func (request *TcpUpdateStreamRequest) Serialize() []byte {
diff --git a/foreign/go/binary_serialization/update_stream_serializer_test.go b/foreign/go/binary_serialization/update_stream_serializer_test.go
index 6b3589a7..d57566cb 100644
--- a/foreign/go/binary_serialization/update_stream_serializer_test.go
+++ b/foreign/go/binary_serialization/update_stream_serializer_test.go
@@ -18,16 +18,15 @@
package binaryserialization
import (
- iggcon "github.com/apache/iggy/foreign/go/contracts"
"testing"
+
+ iggcon "github.com/apache/iggy/foreign/go/contracts"
)
func TestSerialize_UpdateStream(t *testing.T) {
request := TcpUpdateStreamRequest{
- iggcon.UpdateStreamRequest{
- StreamId: iggcon.NewIdentifier("stream"),
- Name: "update_stream",
- },
+ StreamId: iggcon.NewIdentifier("stream"),
+ Name: "update_stream",
}
serialized1 := request.Serialize()
diff --git a/foreign/go/binary_serialization/update_topic_serializer.go b/foreign/go/binary_serialization/update_topic_serializer.go
index 2b31fcd0..20f85f97 100644
--- a/foreign/go/binary_serialization/update_topic_serializer.go
+++ b/foreign/go/binary_serialization/update_topic_serializer.go
@@ -19,15 +19,25 @@ package binaryserialization
import (
"encoding/binary"
+ "time"
iggcon "github.com/apache/iggy/foreign/go/contracts"
)
type TcpUpdateTopicRequest struct {
- iggcon.UpdateTopicRequest
+ StreamId iggcon.Identifier `json:"streamId"`
+ TopicId iggcon.Identifier `json:"topicId"`
+ CompressionAlgorithm uint8 `json:"compressionAlgorithm"`
+ MessageExpiry time.Duration `json:"messageExpiry"`
+ MaxTopicSize uint64 `json:"maxTopicSize"`
+ ReplicationFactor *uint8 `json:"replicationFactor"`
+ Name string `json:"name"`
}
func (request *TcpUpdateTopicRequest) Serialize() []byte {
+ if request.ReplicationFactor == nil {
+ request.ReplicationFactor = new(uint8)
+ }
streamIdBytes := SerializeIdentifier(request.StreamId)
topicIdBytes := SerializeIdentifier(request.TopicId)
@@ -47,7 +57,7 @@ func (request *TcpUpdateTopicRequest) Serialize() []byte {
binary.LittleEndian.PutUint64(buffer[offset:], uint64(request.MaxTopicSize))
offset += 8
- buffer[offset] = request.ReplicationFactor
+ buffer[offset] = *request.ReplicationFactor
offset++
buffer[offset] = uint8(len(request.Name))
diff --git a/foreign/go/binary_serialization/update_topic_serializer_test.go b/foreign/go/binary_serialization/update_topic_serializer_test.go
index 45781f68..355cea07 100644
--- a/foreign/go/binary_serialization/update_topic_serializer_test.go
+++ b/foreign/go/binary_serialization/update_topic_serializer_test.go
@@ -25,13 +25,11 @@ import (
func TestSerialize_UpdateTopic(t *testing.T) {
request := TcpUpdateTopicRequest{
- iggcon.UpdateTopicRequest{
- StreamId: iggcon.NewIdentifier("stream"),
- TopicId: iggcon.NewIdentifier(1),
- Name: "update_topic",
- MessageExpiry: 100000,
- MaxTopicSize: 100,
- },
+ StreamId: iggcon.NewIdentifier("stream"),
+ TopicId: iggcon.NewIdentifier(1),
+ Name: "update_topic",
+ MessageExpiry: 100000,
+ MaxTopicSize: 100,
}
serialized1 := request.Serialize()
diff --git a/foreign/go/cli/cli.go b/foreign/go/cli/cli.go
index 24006e5d..289e5574 100644
--- a/foreign/go/cli/cli.go
+++ b/foreign/go/cli/cli.go
@@ -23,8 +23,9 @@ import (
"fmt"
"os"
- . "github.com/apache/iggy/foreign/go"
. "github.com/apache/iggy/foreign/go/contracts"
+ "github.com/apache/iggy/foreign/go/iggycli"
+ "github.com/apache/iggy/foreign/go/tcp"
)
// CLI commands
@@ -176,11 +177,8 @@ func main() {
}
//this is very temporary
- ms := CreateMessageStream()
- _, err := ms.LogIn(LogInRequest{
- Username: "iggy",
- Password: "iggy",
- })
+ cli := CreateClient()
+ _, err := cli.LoginUser("iggy", "iggy")
if err != nil {
fmt.Println(err)
os.Exit(1)
@@ -194,11 +192,8 @@ func main() {
createStreamCmd.PrintDefaults()
os.Exit(1)
}
-
- err := ms.CreateStream(CreateStreamRequest{
- StreamId: cs_streamId,
- Name: cs_name,
- })
+ uintStreamId := uint32(cs_streamId)
+ _, err = cli.CreateStream(cs_name, &uintStreamId)
if err != nil {
HandleError(err)
}
@@ -211,10 +206,7 @@ func main() {
os.Exit(1)
}
- err := ms.UpdateStream(UpdateStreamRequest{
- StreamId: NewIdentifier(us_streamId),
- Name: us_name,
- })
+ err := cli.UpdateStream(NewIdentifier(us_streamId), us_name)
if err != nil {
HandleError(err)
}
@@ -222,7 +214,7 @@ func main() {
case "getstream":
_ = getStreamCmd.Parse(os.Args[2:])
if gs_streamId == -1 {
- streams, err := ms.GetStreams()
+ streams, err := cli.GetStreams()
if err != nil {
HandleError(err)
}
@@ -230,10 +222,7 @@ func main() {
return
}
- stream, err := ms.GetStreamById(
- GetStreamRequest{
- StreamID: NewIdentifier(gs_streamId),
- })
+ stream, err := cli.GetStream(NewIdentifier(gs_streamId))
if err != nil {
HandleError(err)
}
@@ -247,7 +236,7 @@ func main() {
os.Exit(1)
}
- err := ms.DeleteStream(NewIdentifier(ds_streamId))
+ err := cli.DeleteStream(NewIdentifier(ds_streamId))
if err != nil {
HandleError(err)
}
@@ -260,12 +249,15 @@ func main() {
os.Exit(1)
}
- err := ms.CreateTopic(CreateTopicRequest{
- TopicId: ct_topicId,
- Name: ct_name,
- PartitionsCount: ct_partitionsCount,
- StreamId: NewIdentifier(ct_streamId),
- })
+ _, err := cli.CreateTopic(
+ NewIdentifier(ct_streamId),
+ ct_name,
+ ct_partitionsCount,
+ 0,
+ 0,
+ 0,
+ nil,
+ &ct_topicId)
if err != nil {
HandleError(err)
}
@@ -277,11 +269,15 @@ func main() {
os.Exit(1)
}
- err := ms.UpdateTopic(UpdateTopicRequest{
- TopicId: NewIdentifier(ut_topicId),
- Name: ut_name,
- StreamId: NewIdentifier(ut_streamId),
- })
+ err := cli.UpdateTopic(
+ NewIdentifier(ut_streamId),
+ NewIdentifier(ut_topicId),
+ ut_name,
+ 0,
+ 0,
+ 0,
+ new(uint8),
+ )
if err != nil {
HandleError(err)
}
@@ -290,14 +286,14 @@ func main() {
_ = getTopicCmd.Parse(os.Args[2:])
if gt_topicId == -1 {
- topics, err := ms.GetTopics(NewIdentifier(gt_streamId))
+ topics, err := cli.GetTopics(NewIdentifier(gt_streamId))
if err != nil {
HandleError(err)
}
SerializeAndPrint(topics)
return
}
- topic, err := ms.GetTopicById(NewIdentifier(gt_streamId), NewIdentifier(gt_topicId))
+ topic, err := cli.GetTopic(NewIdentifier(gt_streamId), NewIdentifier(gt_topicId))
if err != nil {
HandleError(err)
}
@@ -317,14 +313,14 @@ func main() {
os.Exit(1)
}
- err := ms.DeleteTopic(NewIdentifier(dt_streamId), NewIdentifier(dt_topicId))
+ err := cli.DeleteTopic(NewIdentifier(dt_streamId), NewIdentifier(dt_topicId))
if err != nil {
HandleError(err)
}
case "getstats":
_ = getStatsCmd.Parse(os.Args[2:])
- stats, err := ms.GetStats()
+ stats, err := cli.GetStats()
if err != nil {
HandleError(err)
}
@@ -339,18 +335,12 @@ func main() {
}
}
-func CreateMessageStream() MessageStream {
- factory := &IggyClientFactory{}
- config := IggyConfiguration{
- BaseAddress: url + ":" + port,
- Protocol: Tcp,
- }
-
- ms, err := factory.CreateMessageStream(config)
+func CreateClient() iggycli.Client {
+ cli, err := iggycli.NewIggyClient(iggycli.WithTcp(tcp.WithServerAddress(url + ":" + port)))
if err != nil {
panic(err)
}
- return ms
+ return cli
}
func SerializeAndPrint(obj any) {
diff --git a/foreign/go/contracts/access_tokens.go b/foreign/go/contracts/access_tokens.go
index 40e2c20c..773a5d77 100644
--- a/foreign/go/contracts/access_tokens.go
+++ b/foreign/go/contracts/access_tokens.go
@@ -19,20 +19,20 @@ package iggcon
import "time"
-type CreateAccessTokenRequest struct {
+type CreatePersonalAccessTokenRequest struct {
Name string `json:"Name"`
Expiry uint32 `json:"Expiry"`
}
-type DeleteAccessTokenRequest struct {
+type DeletePersonalAccessTokenRequest struct {
Name string `json:"Name"`
}
-type AccessTokenResponse struct {
+type PersonalAccessTokenInfo struct {
Name string `json:"Name"`
Expiry *time.Time `json:"Expiry"`
}
-type AccessToken struct {
+type RawPersonalAccessToken struct {
Token string `json:"token"`
}
diff --git a/foreign/go/contracts/client.go b/foreign/go/contracts/client.go
index 581b9d23..265990ff 100644
--- a/foreign/go/contracts/client.go
+++ b/foreign/go/contracts/client.go
@@ -17,11 +17,15 @@
package iggcon
-type ClientResponse struct {
- ID uint32 `json:"id"`
- Address string `json:"address"`
- UserID uint32 `json:"userId"`
- Transport string `json:"transport"`
- ConsumerGroupsCount uint32 `json:"consumerGroupsCount"`
- ConsumerGroups []ConsumerGroupInfo `json:"consumerGroups,omitempty"`
+type ClientInfoDetails struct {
+ ClientInfo
+ ConsumerGroups []ConsumerGroupInfo `json:"consumerGroups,omitempty"`
+}
+
+type ClientInfo struct {
+ ID uint32 `json:"id"`
+ Address string `json:"address"`
+ UserID uint32 `json:"userId"`
+ Transport string `json:"transport"`
+ ConsumerGroupsCount uint32 `json:"consumerGroupsCount"`
}
diff --git a/foreign/go/contracts/config.go b/foreign/go/contracts/config.go
index b25f462d..0098acd6 100644
--- a/foreign/go/contracts/config.go
+++ b/foreign/go/contracts/config.go
@@ -17,11 +17,6 @@
package iggcon
-import (
- "context"
- "time"
-)
-
type IggyMessageCompression string
const (
@@ -32,14 +27,6 @@ const (
// MESSAGE_COMPRESSION_ZSTD IggyMessageCompression = "zstd"
)
-type IggyConfiguration struct {
- context.Context
- BaseAddress string `json:"baseAddress"`
- Protocol Protocol `json:"protocol"`
- MessageCompression IggyMessageCompression `json:"compression"`
- HeartbeatInterval time.Duration `json:"heartbeatInterval"`
-}
-
type Protocol string
const (
diff --git a/foreign/go/contracts/consumer.go b/foreign/go/contracts/consumer.go
index ff3a06c7..af49df1d 100644
--- a/foreign/go/contracts/consumer.go
+++ b/foreign/go/contracts/consumer.go
@@ -20,8 +20,8 @@ package iggcon
type ConsumerKind int
const (
- ConsumerSingle ConsumerKind = 1
- ConsumerGroup ConsumerKind = 2
+ ConsumerKindSingle ConsumerKind = 1
+ ConsumerKindGroup ConsumerKind = 2
)
type Consumer struct {
diff --git a/foreign/go/contracts/consumer_groups.go b/foreign/go/contracts/consumer_groups.go
index 475dea66..f8b6765e 100644
--- a/foreign/go/contracts/consumer_groups.go
+++ b/foreign/go/contracts/consumer_groups.go
@@ -17,17 +17,28 @@
package iggcon
-type ConsumerGroupResponse struct {
+type ConsumerGroup struct {
Id int `json:"id"`
Name string `json:"name"`
PartitionsCount int `json:"partitionsCount"`
MembersCount int `json:"membersCount"`
}
+type ConsumerGroupDetails struct {
+ ConsumerGroup
+ Members []ConsumerGroupMember
+}
+
+type ConsumerGroupMember struct {
+ ID int
+ PartitionsCount int
+ Partitions []int
+}
+
type CreateConsumerGroupRequest struct {
StreamId Identifier `json:"streamId"`
TopicId Identifier `json:"topicId"`
- ConsumerGroupId int `json:"consumerGroupId"`
+ ConsumerGroupId *uint32 `json:"consumerGroupId"`
Name string `json:"name"`
}
diff --git a/foreign/go/contracts/login.go b/foreign/go/contracts/login.go
index bb87b7c7..31c60b9e 100644
--- a/foreign/go/contracts/login.go
+++ b/foreign/go/contracts/login.go
@@ -17,17 +17,20 @@
package iggcon
-type LogInRequest struct {
+type LoginUserRequest struct {
Username string `json:"username"`
Password string `json:"password"`
Version string `json:"version,omitempty"`
Context string `json:"context,omitempty"`
}
-type LogInAccessTokenRequest struct {
+type LoginWithPersonalAccessTokenRequest struct {
Token string `json:"token"`
}
-type LogInResponse struct {
+type IdentityInfo struct {
+ // Unique identifier (numeric) of the user.
UserId uint32 `json:"userId"`
+ // The optional tokens, used only by HTTP transport.
+ AccessToken *string `json:"accessToken"`
}
diff --git a/foreign/go/contracts/messages.go b/foreign/go/contracts/messages.go
index b6dde047..b20e9b11 100644
--- a/foreign/go/contracts/messages.go
+++ b/foreign/go/contracts/messages.go
@@ -45,7 +45,7 @@ const (
MaxUserHeadersSize = 100 * 1000
)
-type FetchMessagesRequest struct {
+type PollMessageRequest struct {
StreamId Identifier `json:"streamId"`
TopicId Identifier `json:"topicId"`
Consumer Consumer `json:"consumer"`
@@ -55,7 +55,7 @@ type FetchMessagesRequest struct {
AutoCommit bool `json:"autoCommit"`
}
-type FetchMessagesResponse struct {
+type PolledMessage struct {
PartitionId uint32
CurrentOffset uint64
MessageCount uint32
diff --git a/foreign/go/contracts/offets.go b/foreign/go/contracts/offets.go
index 21fc3c1f..2522df88 100644
--- a/foreign/go/contracts/offets.go
+++ b/foreign/go/contracts/offets.go
@@ -17,22 +17,22 @@
package iggcon
-type StoreOffsetRequest struct {
+type StoreConsumerOffsetRequest struct {
StreamId Identifier `json:"streamId"`
TopicId Identifier `json:"topicId"`
Consumer Consumer `json:"consumer"`
- PartitionId int `json:"partitionId"`
+ PartitionId *uint32 `json:"partitionId"`
Offset uint64 `json:"offset"`
}
-type GetOffsetRequest struct {
+type GetConsumerOffsetRequest struct {
StreamId Identifier `json:"streamId"`
TopicId Identifier `json:"topicId"`
Consumer Consumer `json:"consumer"`
- PartitionId int `json:"partitionId"`
+ PartitionId *uint32 `json:"partitionId"`
}
-type OffsetResponse struct {
+type ConsumerOffsetInfo struct {
PartitionId int `json:"partitionId"`
CurrentOffset uint64 `json:"currentOffset"`
StoredOffset uint64 `json:"storedOffset"`
diff --git a/foreign/go/contracts/partitions.go b/foreign/go/contracts/partitions.go
index 97058ae5..76d1f80c 100644
--- a/foreign/go/contracts/partitions.go
+++ b/foreign/go/contracts/partitions.go
@@ -36,13 +36,13 @@ type PartitionContract struct {
type CreatePartitionsRequest struct {
StreamId Identifier `json:"streamId"`
TopicId Identifier `json:"topicId"`
- PartitionsCount int `json:"partitionsCount"`
+ PartitionsCount uint32 `json:"partitionsCount"`
}
-type DeletePartitionRequest struct {
+type DeletePartitionsRequest struct {
StreamId Identifier `json:"streamId"`
TopicId Identifier `json:"topicId"`
- PartitionsCount int `json:"partitionsCount"`
+ PartitionsCount uint32 `json:"partitionsCount"`
}
type PartitioningKind int
diff --git a/foreign/go/contracts/stream.go b/foreign/go/contracts/stream.go
index 3a9c8297..8f760ebd 100644
--- a/foreign/go/contracts/stream.go
+++ b/foreign/go/contracts/stream.go
@@ -22,19 +22,18 @@ type CreateStreamRequest struct {
Name string `json:"name"`
}
-type UpdateStreamRequest struct {
- StreamId Identifier `json:"streamId"`
- Name string `json:"name"`
+type Stream struct {
+ Id int `json:"id"`
+ Name string `json:"name"`
+ SizeBytes uint64 `json:"sizeBytes"`
+ CreatedAt uint64 `json:"createdAt"`
+ MessagesCount uint64 `json:"messagesCount"`
+ TopicsCount int `json:"topicsCount"`
}
-type StreamResponse struct {
- Id int `json:"id"`
- Name string `json:"name"`
- SizeBytes uint64 `json:"sizeBytes"`
- CreatedAt uint64 `json:"createdAt"`
- MessagesCount uint64 `json:"messagesCount"`
- TopicsCount int `json:"topicsCount"`
- Topics []TopicResponse `json:"topics,omitempty"`
+type StreamDetails struct {
+ Stream
+ Topics []Topic `json:"topics,omitempty"`
}
type GetStreamRequest struct {
diff --git a/foreign/go/contracts/topics.go b/foreign/go/contracts/topics.go
index eceb2663..504a9e59 100644
--- a/foreign/go/contracts/topics.go
+++ b/foreign/go/contracts/topics.go
@@ -40,16 +40,20 @@ type UpdateTopicRequest struct {
Name string `json:"name"`
}
-type TopicResponse struct {
- Id int `json:"id"`
- CreatedAt int `json:"createdAt"`
- Name string `json:"name"`
- SizeBytes uint64 `json:"sizeBytes"`
- MessageExpiry time.Duration `json:"messageExpiry"`
- CompressionAlgorithm uint8 `json:"compressionAlgorithm"`
- MaxTopicSize uint64 `json:"maxTopicSize"`
- ReplicationFactor uint8 `json:"replicationFactor"`
- MessagesCount uint64 `json:"messagesCount"`
- PartitionsCount int `json:"partitionsCount"`
- Partitions []PartitionContract `json:"partitions,omitempty"`
+type Topic struct {
+ Id int `json:"id"`
+ CreatedAt int `json:"createdAt"`
+ Name string `json:"name"`
+ SizeBytes uint64 `json:"sizeBytes"`
+ MessageExpiry time.Duration `json:"messageExpiry"`
+ CompressionAlgorithm uint8 `json:"compressionAlgorithm"`
+ MaxTopicSize uint64 `json:"maxTopicSize"`
+ ReplicationFactor uint8 `json:"replicationFactor"`
+ MessagesCount uint64 `json:"messagesCount"`
+ PartitionsCount int `json:"partitionsCount"`
+}
+
+type TopicDetails struct {
+ Topic
+ Partitions []PartitionContract `json:"partitions,omitempty"`
}
diff --git a/foreign/go/contracts/users.go b/foreign/go/contracts/users.go
index e6e68262..3fcc051c 100644
--- a/foreign/go/contracts/users.go
+++ b/foreign/go/contracts/users.go
@@ -23,14 +23,14 @@ type ChangePasswordRequest struct {
NewPassword string `json:"NewPassword"`
}
-type UpdateUserPermissionsRequest struct {
+type UpdatePermissionsRequest struct {
UserID Identifier `json:"-"`
Permissions *Permissions `json:"Permissions,omitempty"`
}
type UpdateUserRequest struct {
UserID Identifier `json:"-"`
- Username string `json:"username"`
+ Username *string `json:"username"`
Status *UserStatus `json:"userStatus"`
}
@@ -41,11 +41,15 @@ type CreateUserRequest struct {
Permissions *Permissions `json:"Permissions,omitempty"`
}
-type UserResponse struct {
- Id uint32 `json:"Id"`
- CreatedAt uint64 `json:"CreatedAt"`
- Status UserStatus `json:"Status"`
- Username string `json:"Username"`
+type UserInfo struct {
+ Id uint32 `json:"Id"`
+ CreatedAt uint64 `json:"CreatedAt"`
+ Status UserStatus `json:"Status"`
+ Username string `json:"Username"`
+}
+
+type UserInfoDetails struct {
+ UserInfo
Permissions *Permissions `json:"Permissions"`
}
diff --git a/foreign/go/errors/constants.go b/foreign/go/errors/constants.go
index 2de4ed94..62cb0486 100644
--- a/foreign/go/errors/constants.go
+++ b/foreign/go/errors/constants.go
@@ -22,6 +22,10 @@ var (
Code: 20,
Message: "resource_not_found",
}
+ InvalidConfiguration = &IggyError{
+ Code: 2,
+ Message: "invalid_configuration",
+ }
StreamIdNotFound = &IggyError{
Code: 1009,
Message: "stream_id_not_found",
@@ -30,6 +34,10 @@ var (
Code: 2010,
Message: "topic_id_not_found",
}
+ InvalidMessagesCount = &IggyError{
+ Code: 4009,
+ Message: "invalid_messages_count",
+ }
InvalidMessagePayloadLength = &IggyError{
Code: 4025,
Message: "invalid_message_payload_length",
diff --git a/foreign/go/iggycli/client.go b/foreign/go/iggycli/client.go
new file mode 100644
index 00000000..0080dfc9
--- /dev/null
+++ b/foreign/go/iggycli/client.go
@@ -0,0 +1,260 @@
+// 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 iggycli
+
+import (
+ "time"
+
+ . "github.com/apache/iggy/foreign/go/contracts"
+)
+
+type Client interface {
+ // GetStream get the info about a specific stream by unique ID or name.
+ // Authentication is required, and the permission to read the streams.
+ GetStream(streamId Identifier) (*StreamDetails, error)
+
+ // GetStreams get the info about all the streams.
+ // Authentication is required, and the permission to read the streams.
+ GetStreams() ([]Stream, error)
+
+ // CreateStream create a new stream.
+ // Authentication is required, and the permission to manage the streams.
+ CreateStream(name string, streamId *uint32) (*StreamDetails, error)
+
+ // UpdateStream update a stream by unique ID or name.
+ // Authentication is required, and the permission to manage the streams.
+ UpdateStream(streamId Identifier, name string) error
+
+ // DeleteStream delete a topic by unique ID or name.
+ // Authentication is required, and the permission to manage the topics.
+ DeleteStream(id Identifier) error
+
+ // GetTopic Get the info about a specific topic by unique ID or name.
+ // Authentication is required, and the permission to read the topics.
+ GetTopic(streamId, topicId Identifier) (*TopicDetails, error)
+
+ // GetTopics get the info about all the topics.
+ // Authentication is required, and the permission to read the topics.
+ GetTopics(streamId Identifier) ([]Topic, error)
+
+ // CreateTopic create a new topic.
+ // Authentication is required, and the permission to manage the topics.
+ CreateTopic(
+ streamId Identifier,
+ name string,
+ partitionsCount int,
+ compressionAlgorithm uint8,
+ messageExpiry time.Duration,
+ maxTopicSize uint64,
+ replicationFactor *uint8,
+ topicId *int,
+ ) (*TopicDetails, error)
+
+ // UpdateTopic update a topic by unique ID or name.
+ // Authentication is required, and the permission to manage the topics.
+ UpdateTopic(
+ streamId Identifier,
+ topicId Identifier,
+ name string,
+ compressionAlgorithm uint8,
+ messageExpiry time.Duration,
+ maxTopicSize uint64,
+ replicationFactor *uint8,
+ ) error
+
+ // DeleteTopic delete a topic by unique ID or name.
+ // Authentication is required, and the permission to manage the topics.
+ DeleteTopic(streamId, topicId Identifier) error
+
+ // SendMessages sends messages using specified partitioning strategy to the given stream and topic by unique IDs or names.
+ // Authentication is required, and the permission to send the messages.
+ SendMessages(
+ streamId Identifier,
+ topicId Identifier,
+ partitioning Partitioning,
+ messages []IggyMessage,
+ ) error
+
+ // PollMessages poll given amount of messages using the specified consumer and strategy from the specified stream and topic by unique IDs or names.
+ // Authentication is required, and the permission to poll the messages.
+ PollMessages(
+ streamId Identifier,
+ topicId Identifier,
+ consumer Consumer,
+ strategy PollingStrategy,
+ count uint32,
+ autoCommit bool,
+ partitionId *uint32,
+ ) (*PolledMessage, error)
+
+ // StoreConsumerOffset store the consumer offset for a specific consumer or consumer group for the given stream and topic by unique IDs or names.
+ // Authentication is required, and the permission to poll the messages.
+ StoreConsumerOffset(
+ consumer Consumer,
+ streamId Identifier,
+ topicId Identifier,
+ offset uint64,
+ partitionId *uint32,
+ ) error
+
+ // GetConsumerOffset get the consumer offset for a specific consumer or consumer group for the given stream and topic by unique IDs or names.
+ // Authentication is required, and the permission to poll the messages.
+ GetConsumerOffset(
+ consumer Consumer,
+ streamId Identifier,
+ topicId Identifier,
+ partitionId *uint32,
+ ) (*ConsumerOffsetInfo, error)
+
+ // GetConsumerGroups get the info about all the consumer groups for the given stream and topic by unique IDs or names.
+ // Authentication is required, and the permission to read the streams or topics.
+ GetConsumerGroups(streamId Identifier, topicId Identifier) ([]ConsumerGroup, error)
+
+ // GetConsumerGroup get the info about a specific consumer group by unique ID or name for the given stream and topic by unique IDs or names.
+ // Authentication is required, and the permission to read the streams or topics.
+ GetConsumerGroup(
+ streamId Identifier,
+ topicId Identifier,
+ groupId Identifier,
+ ) (*ConsumerGroupDetails, error)
+
+ // CreateConsumerGroup create a new consumer group for the given stream and topic by unique IDs or names.
+ // Authentication is required, and the permission to manage the streams or topics.
+ CreateConsumerGroup(
+ streamId Identifier,
+ topicId Identifier,
+ name string,
+ groupId *uint32,
+ ) (*ConsumerGroupDetails, error)
+
+ // DeleteConsumerGroup delete a consumer group by unique ID or name for the given stream and topic by unique IDs or names.
+ // Authentication is required, and the permission to manage the streams or topics.
+ DeleteConsumerGroup(
+ streamId Identifier,
+ topicId Identifier,
+ groupId Identifier,
+ ) error
+
+ // JoinConsumerGroup join a consumer group by unique ID or name for the given stream and topic by unique IDs or names.
+ // Authentication is required, and the permission to read the streams or topics.
+ JoinConsumerGroup(
+ streamId Identifier,
+ topicId Identifier,
+ groupId Identifier,
+ ) error
+
+ // LeaveConsumerGroup leave a consumer group by unique ID or name for the given stream and topic by unique IDs or names.
+ // Authentication is required, and the permission to read the streams or topics.
+ LeaveConsumerGroup(
+ streamId Identifier,
+ topicId Identifier,
+ groupId Identifier,
+ ) error
+
+ // CreatePartitions create new N partitions for a topic by unique ID or name.
+ // For example, given a topic with 3 partitions, if you create 2 partitions, the topic will have 5 partitions (from 1 to 5).
+ // Authentication is required, and the permission to manage the partitions.
+ CreatePartitions(
+ streamId Identifier,
+ topicId Identifier,
+ partitionsCount uint32,
+ ) error
+
+ // DeletePartitions delete last N partitions for a topic by unique ID or name.
+ // For example, given a topic with 5 partitions, if you delete 2 partitions, the topic will have 3 partitions left (from 1 to 3).
+ // Authentication is required, and the permission to manage the partitions.
+ DeletePartitions(
+ streamId Identifier,
+ topicId Identifier,
+ partitionsCount uint32,
+ ) error
+
+ // GetUser get the info about a specific user by unique ID or username.
+ // Authentication is required, and the permission to read the users, unless the provided user ID is the same as the authenticated user.
+ GetUser(identifier Identifier) (*UserInfoDetails, error)
+
+ // GetUsers get the info about all the users.
+ // Authentication is required, and the permission to read the users.
+ GetUsers() ([]UserInfo, error)
+
+ // CreateUser create a new user.
+ // Authentication is required, and the permission to manage the users.
+ CreateUser(
+ username string,
+ password string,
+ status UserStatus,
+ permissions *Permissions,
+ ) (*UserInfoDetails, error)
+
+ // UpdateUser update a user by unique ID or username.
+ // Authentication is required, and the permission to manage the users.
+ UpdateUser(
+ userID Identifier,
+ username *string,
+ status *UserStatus,
+ ) error
+
+ // UpdatePermissions update the permissions of a user by unique ID or username.
+ // Authentication is required, and the permission to manage the users.
+ UpdatePermissions(userID Identifier, permissions *Permissions) error
+
+ // ChangePassword change the password of a user by unique ID or username.
+ // Authentication is required, and the permission to manage the users, unless the provided user ID is the same as the authenticated user.
+ ChangePassword(
+ userID Identifier,
+ currentPassword string,
+ newPassword string,
+ ) error
+
+ // DeleteUser delete a user by unique ID or username.
+ // Authentication is required, and the permission to manage the users.
+ DeleteUser(identifier Identifier) error
+
+ // CreatePersonalAccessToken create a new personal access token for the currently authenticated user.
+ CreatePersonalAccessToken(name string, expiry uint32) (*RawPersonalAccessToken, error)
+
+ // DeletePersonalAccessToken delete a personal access token of the currently authenticated user by unique token name.
+ DeletePersonalAccessToken(name string) error
+
+ // GetPersonalAccessTokens get the info about all the personal access tokens of the currently authenticated user.
+ GetPersonalAccessTokens() ([]PersonalAccessTokenInfo, error)
+
+ // LoginWithPersonalAccessToken login the user with the provided personal access token.
+ LoginWithPersonalAccessToken(token string) (*IdentityInfo, error)
+
+ // LoginUser login a user by username and password.
+ LoginUser(username string, password string) (*IdentityInfo, error)
+
+ // LogoutUser logout the currently authenticated user.
+ LogoutUser() error
+
+ // GetStats get the stats of the system such as PID, memory usage, streams count etc.
+ // Authentication is required, and the permission to read the server info.
+ GetStats() (*Stats, error)
+
+ // Ping the server to check if it's alive.
+ Ping() error
+
+ // GetClients get the info about all the currently connected clients (not to be confused with the users).
+ // Authentication is required, and the permission to read the server info.
+ GetClients() ([]ClientInfo, error)
+
+ // GetClient get the info about a specific client by unique ID (not to be confused with the user).
+ // Authentication is required, and the permission to read the server info.
+ GetClient(clientId int) (*ClientInfoDetails, error)
+}
diff --git a/foreign/go/iggycli/iggy_client.go b/foreign/go/iggycli/iggy_client.go
new file mode 100644
index 00000000..eafc4e77
--- /dev/null
+++ b/foreign/go/iggycli/iggy_client.go
@@ -0,0 +1,71 @@
+// 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 iggycli
+
+import (
+ "fmt"
+
+ . "github.com/apache/iggy/foreign/go/contracts"
+ "github.com/apache/iggy/foreign/go/tcp"
+)
+
+type Options struct {
+ protocol Protocol
+ tcpOptions []tcp.Option
+}
+
+func GetDefaultOptions() Options {
+ return Options{
+ protocol: Tcp,
+ tcpOptions: nil,
+ }
+}
+
+type Option func(*Options)
+
+// WithTcp sets the client protocol to TCP and applies custom TCP options.
+func WithTcp(tcpOpts ...tcp.Option) Option {
+ return func(opts *Options) {
+ opts.protocol = Tcp
+ opts.tcpOptions = tcpOpts
+ }
+}
+
+// NewIggyClient create the IggyClient instance.
+// If no Option is provided, NewIggyClient will create a default TCP client.
+func NewIggyClient(options ...Option) (Client, error) {
+ opts := GetDefaultOptions()
+
+ for _, opt := range options {
+ opt(&opts)
+ }
+
+ var err error
+ var cli Client
+ switch opts.protocol {
+ case Tcp:
+ cli, err = tcp.NewIggyTcpClient(opts.tcpOptions...)
+ default:
+ return nil, fmt.Errorf("unknown protocol type: %v", opts.protocol)
+ }
+ if err != nil {
+ return nil, fmt.Errorf("failed to create an iggy client: %w", err)
+ }
+
+ return cli, nil
+}
diff --git a/foreign/go/message_stream.go b/foreign/go/message_stream.go
deleted file mode 100644
index 706512ad..00000000
--- a/foreign/go/message_stream.go
+++ /dev/null
@@ -1,72 +0,0 @@
-// 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 iggy
-
-import . "github.com/apache/iggy/foreign/go/contracts"
-
-type MessageStream interface {
- GetStreamById(request GetStreamRequest) (*StreamResponse, error)
- GetStreams() ([]StreamResponse, error)
- CreateStream(request CreateStreamRequest) error
- UpdateStream(request UpdateStreamRequest) error
- DeleteStream(id Identifier) error
-
- GetTopicById(streamId, topicId Identifier) (*TopicResponse, error)
- GetTopics(streamId Identifier) ([]TopicResponse, error)
- CreateTopic(request CreateTopicRequest) error
- UpdateTopic(request UpdateTopicRequest) error
- DeleteTopic(streamId, topicId Identifier) error
-
- SendMessages(request SendMessagesRequest) error
- PollMessages(request FetchMessagesRequest) (*FetchMessagesResponse, error)
-
- StoreOffset(request StoreOffsetRequest) error
- GetOffset(request GetOffsetRequest) (*OffsetResponse, error)
-
- GetConsumerGroups(streamId Identifier, topicId Identifier) ([]ConsumerGroupResponse, error)
- GetConsumerGroupById(streamId, topicId, groupId Identifier) (*ConsumerGroupResponse, error)
- CreateConsumerGroup(request CreateConsumerGroupRequest) error
- DeleteConsumerGroup(request DeleteConsumerGroupRequest) error
- JoinConsumerGroup(request JoinConsumerGroupRequest) error
- LeaveConsumerGroup(request LeaveConsumerGroupRequest) error
-
- CreatePartition(request CreatePartitionsRequest) error
- DeletePartition(request DeletePartitionRequest) error
-
- GetUser(identifier Identifier) (*UserResponse, error)
- GetUsers() ([]*UserResponse, error)
- CreateUser(request CreateUserRequest) error
- UpdateUser(request UpdateUserRequest) error
- UpdateUserPermissions(request UpdateUserPermissionsRequest) error
- ChangePassword(request ChangePasswordRequest) error
- DeleteUser(identifier Identifier) error
-
- CreateAccessToken(request CreateAccessTokenRequest) (*AccessToken, error)
- DeleteAccessToken(request DeleteAccessTokenRequest) error
- GetAccessTokens() ([]AccessTokenResponse, error)
-
- LogIn(request LogInRequest) (*LogInResponse, error)
- LogInWithAccessToken(request LogInAccessTokenRequest) (*LogInResponse, error)
- LogOut() error
-
- GetStats() (*Stats, error)
- Ping() error
-
- GetClients() ([]ClientResponse, error)
- GetClientById(clientId int) (*ClientResponse, error)
-}
diff --git a/foreign/go/message_stream_factory.go b/foreign/go/message_stream_factory.go
deleted file mode 100644
index ad9aa4e2..00000000
--- a/foreign/go/message_stream_factory.go
+++ /dev/null
@@ -1,55 +0,0 @@
-// 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 iggy
-
-import (
- "context"
- "errors"
-
- "github.com/apache/iggy/foreign/go/tcp"
-
- iggcon "github.com/apache/iggy/foreign/go/contracts"
-)
-
-type IMessageStreamFactory interface {
- CreateStream(config iggcon.IggyConfiguration) (MessageStream, error)
-}
-
-type IggyClientFactory struct{}
-
-func (msf *IggyClientFactory) CreateMessageStream(config iggcon.IggyConfiguration) (MessageStream, error) {
- // Support previous behaviour
- if config.Context == nil {
- config.Context = context.Background()
- }
-
- if config.Protocol == iggcon.Tcp {
- tcpMessageStream, err := tcp.NewTcpMessageStream(
- config.Context,
- config.BaseAddress,
- config.MessageCompression,
- config.HeartbeatInterval,
- )
- if err != nil {
- return nil, err
- }
- return tcpMessageStream, nil
- }
-
- return nil, errors.New("unsupported protocol")
-}
diff --git a/foreign/go/samples/consumer/consumer.go b/foreign/go/samples/consumer/consumer.go
index 9b12c522..4390ff7b 100644
--- a/foreign/go/samples/consumer/consumer.go
+++ b/foreign/go/samples/consumer/consumer.go
@@ -23,9 +23,10 @@ import (
"fmt"
"time"
- . "github.com/apache/iggy/foreign/go"
. "github.com/apache/iggy/foreign/go/contracts"
+ "github.com/apache/iggy/foreign/go/iggycli"
sharedDemoContracts "github.com/apache/iggy/foreign/go/samples/shared"
+ "github.com/apache/iggy/foreign/go/tcp"
)
// config
@@ -38,42 +39,32 @@ const (
)
func main() {
- factory := &IggyClientFactory{}
- config := IggyConfiguration{
- BaseAddress: "127.0.0.1:8090",
- Protocol: Tcp,
- }
-
- messageStream, err := factory.CreateMessageStream(config)
+ cli, err := iggycli.NewIggyClient(
+ iggycli.WithTcp(
+ tcp.WithServerAddress("127.0.0.1:8090"),
+ ),
+ )
if err != nil {
panic(err)
}
-
- _, err = messageStream.LogIn(LogInRequest{
- Username: "iggy",
- Password: "iggy",
- })
+ _, err = cli.LoginUser("iggy", "iggy")
if err != nil {
panic("COULD NOT LOG IN")
}
- if err = EnsureInsfrastructureIsInitialized(messageStream); err != nil {
+ if err = EnsureInfrastructureIsInitialized(cli); err != nil {
panic(err)
}
- if err := ConsumeMessages(messageStream); err != nil {
+ if err = ConsumeMessages(cli); err != nil {
panic(err)
}
}
-func EnsureInsfrastructureIsInitialized(messageStream MessageStream) error {
- if _, streamErr := messageStream.GetStreamById(GetStreamRequest{
- StreamID: NewIdentifier(DefaultStreamId),
- }); streamErr != nil {
- streamErr = messageStream.CreateStream(CreateStreamRequest{
- StreamId: DefaultStreamId,
- Name: "Test Producer Stream",
- })
+func EnsureInfrastructureIsInitialized(cli iggycli.Client) error {
+ if _, streamErr := cli.GetStream(NewIdentifier(DefaultStreamId)); streamErr != nil {
+ uint32DefaultStreamId := uint32(DefaultStreamId)
+ _, streamErr = cli.CreateStream("Test Producer Stream", &uint32DefaultStreamId)
if streamErr != nil {
panic(streamErr)
@@ -84,13 +75,17 @@ func EnsureInsfrastructureIsInitialized(messageStream MessageStream) error {
fmt.Printf("Stream with ID: %d exists.\n", DefaultStreamId)
- if _, topicErr := messageStream.GetTopicById(NewIdentifier(DefaultStreamId), NewIdentifier(TopicId)); topicErr != nil {
- topicErr = messageStream.CreateTopic(CreateTopicRequest{
- TopicId: TopicId,
- Name: "Test Topic From Producer Sample",
- PartitionsCount: 12,
- StreamId: NewIdentifier(DefaultStreamId),
- })
+ if _, topicErr := cli.GetTopic(NewIdentifier(DefaultStreamId), NewIdentifier(TopicId)); topicErr != nil {
+ uint32TopicId := TopicId
+ _, topicErr = cli.CreateTopic(
+ NewIdentifier(DefaultStreamId),
+ "Test Topic From Producer Sample",
+ 12,
+ 0,
+ 0,
+ 0,
+ nil,
+ &uint32TopicId)
if topicErr != nil {
panic(topicErr)
@@ -104,19 +99,19 @@ func EnsureInsfrastructureIsInitialized(messageStream MessageStream) error {
return nil
}
-func ConsumeMessages(messageStream MessageStream) error {
+func ConsumeMessages(cli iggycli.Client) error {
fmt.Printf("Messages will be polled from stream '%d', topic '%d', partition '%d' with interval %d ms.\n", DefaultStreamId, TopicId, Partition, Interval)
for {
- messagesWrapper, err := messageStream.PollMessages(FetchMessagesRequest{
- Count: 1,
- StreamId: NewIdentifier(DefaultStreamId),
- TopicId: NewIdentifier(TopicId),
- Consumer: Consumer{Kind: ConsumerSingle, Id: NewIdentifier(ConsumerId)},
- PartitionId: Partition,
- PollingStrategy: NextPollingStrategy(),
- AutoCommit: true,
- })
+ partionId := uint32(Partition)
+ messagesWrapper, err := cli.PollMessages(
+ NewIdentifier(DefaultStreamId),
+ NewIdentifier(TopicId),
+ Consumer{Kind: ConsumerKindSingle, Id: NewIdentifier(ConsumerId)},
+ NextPollingStrategy(),
+ 1,
+ true,
+ &partionId)
if err != nil {
return err
}
@@ -162,7 +157,7 @@ func HandleMessage(iggyMessage IggyMessage) error {
return err
}
} else {
- if err := json.Unmarshal([]byte(iggyMessage.Payload), &envelope); err != nil {
+ if err := json.Unmarshal(iggyMessage.Payload, &envelope); err != nil {
return err
}
}
diff --git a/foreign/go/samples/producer/producer.go b/foreign/go/samples/producer/producer.go
index 52946642..4135ced1 100644
--- a/foreign/go/samples/producer/producer.go
+++ b/foreign/go/samples/producer/producer.go
@@ -21,9 +21,10 @@ import (
"fmt"
"time"
+ "github.com/apache/iggy/foreign/go/iggycli"
+ "github.com/apache/iggy/foreign/go/tcp"
"github.com/google/uuid"
- . "github.com/apache/iggy/foreign/go"
. "github.com/apache/iggy/foreign/go/contracts"
sharedDemoContracts "github.com/apache/iggy/foreign/go/samples/shared"
)
@@ -37,40 +38,32 @@ const (
)
func main() {
- factory := &IggyClientFactory{}
- config := IggyConfiguration{
- BaseAddress: "127.0.0.1:8090",
- Protocol: Tcp,
- }
-
- messageStream, err := factory.CreateMessageStream(config)
+ cli, err := iggycli.NewIggyClient(
+ iggycli.WithTcp(
+ tcp.WithServerAddress("127.0.0.1:8090"),
+ ),
+ )
if err != nil {
panic(err)
}
- _, err = messageStream.LogIn(LogInRequest{
- Username: "iggy",
- Password: "iggy",
- })
+ _, err = cli.LoginUser("iggy", "iggy")
if err != nil {
panic("COULD NOT LOG IN")
}
- if err = EnsureInsfrastructureIsInitialized(messageStream); err != nil {
+ if err = EnsureInfrastructureIsInitialized(cli); err != nil {
panic(err)
}
- if err = PublishMessages(messageStream); err != nil {
+ if err = PublishMessages(cli); err != nil {
panic(err)
}
}
-func EnsureInsfrastructureIsInitialized(messageStream MessageStream) error {
- if _, streamErr := messageStream.GetStreamById(GetStreamRequest{
- StreamID: NewIdentifier(StreamId)}); streamErr != nil {
- streamErr = messageStream.CreateStream(CreateStreamRequest{
- StreamId: StreamId,
- Name: "Test Producer Stream",
- })
+func EnsureInfrastructureIsInitialized(cli iggycli.Client) error {
+ if _, streamErr := cli.GetStream(NewIdentifier(StreamId)); streamErr != nil {
+ uint32StreamId := uint32(StreamId)
+ _, streamErr = cli.CreateStream("Test Producer Stream", &uint32StreamId)
fmt.Println(StreamId)
@@ -83,13 +76,17 @@ func EnsureInsfrastructureIsInitialized(messageStream MessageStream) error {
fmt.Printf("Stream with ID: %d exists.\n", StreamId)
- if _, topicErr := messageStream.GetTopicById(NewIdentifier(StreamId), NewIdentifier(TopicId)); topicErr != nil {
- topicErr = messageStream.CreateTopic(CreateTopicRequest{
- TopicId: TopicId,
- Name: "Test Topic From Producer Sample",
- PartitionsCount: 12,
- StreamId: NewIdentifier(StreamId),
- })
+ if _, topicErr := cli.GetTopic(NewIdentifier(StreamId), NewIdentifier(TopicId)); topicErr != nil {
+ refStreamId := StreamId
+ _, topicErr = cli.CreateTopic(
+ NewIdentifier(StreamId),
+ "Test Topic From Producer Sample",
+ 12,
+ 0,
+ 0,
+ 0,
+ nil,
+ &refStreamId)
if topicErr != nil {
panic(topicErr)
@@ -103,7 +100,7 @@ func EnsureInsfrastructureIsInitialized(messageStream MessageStream) error {
return nil
}
-func PublishMessages(messageStream MessageStream) error {
+func PublishMessages(messageStream iggycli.Client) error {
fmt.Printf("Messages will be sent to stream '%d', topic '%d', partition '%d' with interval %d ms.\n", StreamId, TopicId, Partition, Interval)
messageGenerator := NewMessageGenerator()
@@ -127,12 +124,12 @@ func PublishMessages(messageStream MessageStream) error {
})
}
- err := messageStream.SendMessages(SendMessagesRequest{
- StreamId: NewIdentifier(StreamId),
- TopicId: NewIdentifier(TopicId),
- Messages: messages,
- Partitioning: PartitionId(Partition),
- })
+ err := messageStream.SendMessages(
+ NewIdentifier(StreamId),
+ NewIdentifier(TopicId),
+ PartitionId(Partition),
+ messages,
+ )
if err != nil {
fmt.Printf("%s", err)
return nil
diff --git a/foreign/go/tcp/tcp_access_token_managament.go b/foreign/go/tcp/tcp_access_token_managament.go
index fb5d1a8c..4ea926c2 100644
--- a/foreign/go/tcp/tcp_access_token_managament.go
+++ b/foreign/go/tcp/tcp_access_token_managament.go
@@ -22,8 +22,11 @@ import (
. "github.com/apache/iggy/foreign/go/contracts"
)
-func (tms *IggyTcpClient) CreateAccessToken(request CreateAccessTokenRequest) (*AccessToken, error) {
- message := binaryserialization.SerializeCreatePersonalAccessToken(request)
+func (tms *IggyTcpClient) CreatePersonalAccessToken(name string, expiry uint32) (*RawPersonalAccessToken, error) {
+ message := binaryserialization.SerializeCreatePersonalAccessToken(CreatePersonalAccessTokenRequest{
+ Name: name,
+ Expiry: expiry,
+ })
buffer, err := tms.sendAndFetchResponse(message, CreateAccessTokenCode)
if err != nil {
return nil, err
@@ -32,13 +35,15 @@ func (tms *IggyTcpClient) CreateAccessToken(request CreateAccessTokenRequest) (*
return binaryserialization.DeserializeAccessToken(buffer)
}
-func (tms *IggyTcpClient) DeleteAccessToken(request DeleteAccessTokenRequest) error {
- message := binaryserialization.SerializeDeletePersonalAccessToken(request)
+func (tms *IggyTcpClient) DeletePersonalAccessToken(name string) error {
+ message := binaryserialization.SerializeDeletePersonalAccessToken(DeletePersonalAccessTokenRequest{
+ Name: name,
+ })
_, err := tms.sendAndFetchResponse(message, DeleteAccessTokenCode)
return err
}
-func (tms *IggyTcpClient) GetAccessTokens() ([]AccessTokenResponse, error) {
+func (tms *IggyTcpClient) GetPersonalAccessTokens() ([]PersonalAccessTokenInfo, error) {
buffer, err := tms.sendAndFetchResponse([]byte{}, GetAccessTokensCode)
if err != nil {
return nil, err
diff --git a/foreign/go/tcp/tcp_clients_managament.go b/foreign/go/tcp/tcp_clients_managament.go
index b655b90b..69cb4caf 100644
--- a/foreign/go/tcp/tcp_clients_managament.go
+++ b/foreign/go/tcp/tcp_clients_managament.go
@@ -22,7 +22,7 @@ import (
. "github.com/apache/iggy/foreign/go/contracts"
)
-func (tms *IggyTcpClient) GetClients() ([]ClientResponse, error) {
+func (tms *IggyTcpClient) GetClients() ([]ClientInfo, error) {
buffer, err := tms.sendAndFetchResponse([]byte{}, GetClientsCode)
if err != nil {
return nil, err
@@ -31,7 +31,7 @@ func (tms *IggyTcpClient) GetClients() ([]ClientResponse, error) {
return binaryserialization.DeserializeClients(buffer)
}
-func (tms *IggyTcpClient) GetClientById(clientId int) (*ClientResponse, error) {
+func (tms *IggyTcpClient) GetClient(clientId int) (*ClientInfoDetails, error) {
message := binaryserialization.SerializeInt(clientId)
buffer, err := tms.sendAndFetchResponse(message, GetClientCode)
if err != nil {
diff --git a/foreign/go/tcp/tcp_consumer_group_managament.go b/foreign/go/tcp/tcp_consumer_group_managament.go
index 9fbf87f1..8b0f4468 100644
--- a/foreign/go/tcp/tcp_consumer_group_managament.go
+++ b/foreign/go/tcp/tcp_consumer_group_managament.go
@@ -23,7 +23,7 @@ import (
ierror "github.com/apache/iggy/foreign/go/errors"
)
-func (tms *IggyTcpClient) GetConsumerGroups(streamId, topicId Identifier) ([]ConsumerGroupResponse, error) {
+func (tms *IggyTcpClient) GetConsumerGroups(streamId, topicId Identifier) ([]ConsumerGroup, error) {
message := binaryserialization.SerializeIdentifiers(streamId, topicId)
buffer, err := tms.sendAndFetchResponse(message, GetGroupsCode)
if err != nil {
@@ -33,7 +33,7 @@ func (tms *IggyTcpClient) GetConsumerGroups(streamId, topicId Identifier) ([]Con
return binaryserialization.DeserializeConsumerGroups(buffer), err
}
-func (tms *IggyTcpClient) GetConsumerGroupById(streamId, topicId, groupId Identifier) (*ConsumerGroupResponse, error) {
+func (tms *IggyTcpClient) GetConsumerGroup(streamId, topicId, groupId Identifier) (*ConsumerGroupDetails, error) {
message := binaryserialization.SerializeIdentifiers(streamId, topicId, groupId)
buffer, err := tms.sendAndFetchResponse(message, GetGroupCode)
if err != nil {
@@ -43,32 +43,42 @@ func (tms *IggyTcpClient) GetConsumerGroupById(streamId, topicId, groupId Identi
return nil, ierror.ConsumerGroupIdNotFound
}
- return binaryserialization.DeserializeConsumerGroup(buffer)
+ consumerGroupDetails := binaryserialization.DeserializeConsumerGroup(buffer)
+ return consumerGroupDetails, err
}
-func (tms *IggyTcpClient) CreateConsumerGroup(request CreateConsumerGroupRequest) error {
- if MaxStringLength < len(request.Name) {
- return ierror.TextTooLong("consumer_group_name")
+func (tms *IggyTcpClient) CreateConsumerGroup(streamId Identifier, topicId Identifier, name string, groupId *uint32) (*ConsumerGroupDetails, error) {
+ if MaxStringLength < len(name) {
+ return nil, ierror.TextTooLong("consumer_group_name")
}
- message := binaryserialization.CreateGroup(request)
- _, err := tms.sendAndFetchResponse(message, CreateGroupCode)
- return err
+ message := binaryserialization.CreateGroup(CreateConsumerGroupRequest{
+ StreamId: streamId,
+ TopicId: topicId,
+ ConsumerGroupId: groupId,
+ Name: name,
+ })
+ buffer, err := tms.sendAndFetchResponse(message, CreateGroupCode)
+ if err != nil {
+ return nil, err
+ }
+ consumerGroup := binaryserialization.DeserializeConsumerGroup(buffer)
+ return consumerGroup, err
}
-func (tms *IggyTcpClient) DeleteConsumerGroup(request DeleteConsumerGroupRequest) error {
- message := binaryserialization.SerializeIdentifiers(request.StreamId, request.TopicId, request.ConsumerGroupId)
+func (tms *IggyTcpClient) DeleteConsumerGroup(streamId Identifier, topicId Identifier, groupId Identifier) error {
+ message := binaryserialization.SerializeIdentifiers(streamId, topicId, groupId)
_, err := tms.sendAndFetchResponse(message, DeleteGroupCode)
return err
}
-func (tms *IggyTcpClient) JoinConsumerGroup(request JoinConsumerGroupRequest) error {
- message := binaryserialization.SerializeIdentifiers(request.StreamId, request.TopicId, request.ConsumerGroupId)
+func (tms *IggyTcpClient) JoinConsumerGroup(streamId Identifier, topicId Identifier, groupId Identifier) error {
+ message := binaryserialization.SerializeIdentifiers(streamId, topicId, groupId)
_, err := tms.sendAndFetchResponse(message, JoinGroupCode)
return err
}
-func (tms *IggyTcpClient) LeaveConsumerGroup(request LeaveConsumerGroupRequest) error {
- message := binaryserialization.SerializeIdentifiers(request.StreamId, request.TopicId, request.ConsumerGroupId)
+func (tms *IggyTcpClient) LeaveConsumerGroup(streamId Identifier, topicId Identifier, groupId Identifier) error {
+ message := binaryserialization.SerializeIdentifiers(streamId, topicId, groupId)
_, err := tms.sendAndFetchResponse(message, LeaveGroupCode)
return err
}
diff --git a/foreign/go/tcp/tcp_core.go b/foreign/go/tcp/tcp_core.go
index 5331f99e..94f5135e 100644
--- a/foreign/go/tcp/tcp_core.go
+++ b/foreign/go/tcp/tcp_core.go
@@ -29,29 +29,54 @@ import (
ierror "github.com/apache/iggy/foreign/go/errors"
)
+type Option func(config *Options)
+
+type Options struct {
+ Ctx context.Context
+ ServerAddress string
+ HeartbeatInterval time.Duration
+}
+
+func GetDefaultOptions() Options {
+ return Options{
+ Ctx: context.Background(),
+ ServerAddress: "127.0.0.1:8090",
+ HeartbeatInterval: time.Second * 5,
+ }
+}
+
type IggyTcpClient struct {
- client *net.TCPConn
+ conn *net.TCPConn
mtx sync.Mutex
MessageCompression iggcon.IggyMessageCompression
}
-const (
- InitialBytesLength = 4
- ExpectedResponseSize = 8
- MaxStringLength = 255
-)
+// WithServerAddress Sets the server address for the TCP client.
+func WithServerAddress(address string) Option {
+ return func(opts *Options) {
+ opts.ServerAddress = address
+ }
+}
-func NewTcpMessageStream(
- ctx context.Context,
- url string,
- compression iggcon.IggyMessageCompression,
- heartbeatInterval time.Duration,
-) (*IggyTcpClient, error) {
- addr, err := net.ResolveTCPAddr("tcp", url)
+// WithContext sets context
+func WithContext(ctx context.Context) Option {
+ return func(opts *Options) {
+ opts.Ctx = ctx
+ }
+}
+
+func NewIggyTcpClient(options ...Option) (*IggyTcpClient, error) {
+ opts := GetDefaultOptions()
+ for _, opt := range options {
+ if opt != nil {
+ opt(&opts)
+ }
+ }
+ addr, err := net.ResolveTCPAddr("tcp", opts.ServerAddress)
if err != nil {
return nil, err
}
-
+ ctx := opts.Ctx
var d = net.Dialer{
KeepAlive: -1,
}
@@ -60,8 +85,11 @@ func NewTcpMessageStream(
return nil, err
}
- client := &IggyTcpClient{client: conn.(*net.TCPConn), MessageCompression: compression}
+ client := &IggyTcpClient{
+ conn: conn.(*net.TCPConn),
+ }
+ heartbeatInterval := opts.HeartbeatInterval
if heartbeatInterval > 0 {
go func() {
ticker := time.NewTicker(heartbeatInterval)
@@ -80,13 +108,19 @@ func NewTcpMessageStream(
return client, nil
}
+const (
+ InitialBytesLength = 4
+ ExpectedResponseSize = 8
+ MaxStringLength = 255
+)
+
func (tms *IggyTcpClient) read(expectedSize int) (int, []byte, error) {
var totalRead int
buffer := make([]byte, expectedSize)
for totalRead < expectedSize {
readSize := expectedSize - totalRead
- n, err := tms.client.Read(buffer[totalRead : totalRead+readSize])
+ n, err := tms.conn.Read(buffer[totalRead : totalRead+readSize])
if err != nil {
return totalRead, buffer[:totalRead], err
}
@@ -99,7 +133,7 @@ func (tms *IggyTcpClient) read(expectedSize int) (int, []byte, error) {
func (tms *IggyTcpClient) write(payload []byte) (int, error) {
var totalWritten int
for totalWritten < len(payload) {
- n, err := tms.client.Write(payload[totalWritten:])
+ n, err := tms.conn.Write(payload[totalWritten:])
if err != nil {
return totalWritten, err
}
diff --git a/foreign/go/tcp/tcp_messaging.go b/foreign/go/tcp/tcp_messaging.go
index 6c36ce7a..26e3a3cf 100644
--- a/foreign/go/tcp/tcp_messaging.go
+++ b/foreign/go/tcp/tcp_messaging.go
@@ -23,17 +23,43 @@ import (
ierror "github.com/apache/iggy/foreign/go/errors"
)
-func (tms *IggyTcpClient) SendMessages(request SendMessagesRequest) error {
- if len(request.Messages) == 0 {
+func (tms *IggyTcpClient) SendMessages(
+ streamId Identifier,
+ topicId Identifier,
+ partitioning Partitioning,
+ messages []IggyMessage,
+) error {
+ if len(messages) == 0 {
return ierror.CustomError("messages_count_should_be_greater_than_zero")
}
- serializedRequest := binaryserialization.TcpSendMessagesRequest{SendMessagesRequest: request}
+ serializedRequest := binaryserialization.TcpSendMessagesRequest{
+ StreamId: streamId,
+ TopicId: topicId,
+ Partitioning: partitioning,
+ Messages: messages,
+ }
_, err := tms.sendAndFetchResponse(serializedRequest.Serialize(tms.MessageCompression), SendMessagesCode)
return err
}
-func (tms *IggyTcpClient) PollMessages(request FetchMessagesRequest) (*FetchMessagesResponse, error) {
- serializedRequest := binaryserialization.TcpFetchMessagesRequest{FetchMessagesRequest: request}
+func (tms *IggyTcpClient) PollMessages(
+ streamId Identifier,
+ topicId Identifier,
+ consumer Consumer,
+ strategy PollingStrategy,
+ count uint32,
+ autoCommit bool,
+ partitionId *uint32,
+) (*PolledMessage, error) {
+ serializedRequest := binaryserialization.TcpFetchMessagesRequest{
+ StreamId: streamId,
+ TopicId: topicId,
+ Consumer: consumer,
+ AutoCommit: autoCommit,
+ Strategy: strategy,
+ Count: count,
+ PartitionId: partitionId,
+ }
buffer, err := tms.sendAndFetchResponse(serializedRequest.Serialize(), PollMessagesCode)
if err != nil {
return nil, err
diff --git a/foreign/go/tcp/tcp_offset_managament.go b/foreign/go/tcp/tcp_offset_managament.go
index 6f1f172d..3e04d0bc 100644
--- a/foreign/go/tcp/tcp_offset_managament.go
+++ b/foreign/go/tcp/tcp_offset_managament.go
@@ -22,8 +22,13 @@ import (
. "github.com/apache/iggy/foreign/go/contracts"
)
-func (tms *IggyTcpClient) GetOffset(request GetOffsetRequest) (*OffsetResponse, error) {
- message := binaryserialization.GetOffset(request)
+func (tms *IggyTcpClient) GetConsumerOffset(consumer Consumer, streamId Identifier, topicId Identifier, partitionId *uint32) (*ConsumerOffsetInfo, error) {
+ message := binaryserialization.GetOffset(GetConsumerOffsetRequest{
+ StreamId: streamId,
+ TopicId: topicId,
+ Consumer: consumer,
+ PartitionId: partitionId,
+ })
buffer, err := tms.sendAndFetchResponse(message, GetOffsetCode)
if err != nil {
return nil, err
@@ -32,8 +37,14 @@ func (tms *IggyTcpClient) GetOffset(request GetOffsetRequest) (*OffsetResponse,
return binaryserialization.DeserializeOffset(buffer), nil
}
-func (tms *IggyTcpClient) StoreOffset(request StoreOffsetRequest) error {
- message := binaryserialization.UpdateOffset(request)
+func (tms *IggyTcpClient) StoreConsumerOffset(consumer Consumer, streamId Identifier, topicId Identifier, offset uint64, partitionId *uint32) error {
+ message := binaryserialization.UpdateOffset(StoreConsumerOffsetRequest{
+ StreamId: streamId,
+ TopicId: topicId,
+ Offset: offset,
+ Consumer: consumer,
+ PartitionId: partitionId,
+ })
_, err := tms.sendAndFetchResponse(message, StoreOffsetCode)
return err
}
diff --git a/foreign/go/tcp/tcp_partition_managament.go b/foreign/go/tcp/tcp_partition_managament.go
index f7af8041..943feaaa 100644
--- a/foreign/go/tcp/tcp_partition_managament.go
+++ b/foreign/go/tcp/tcp_partition_managament.go
@@ -22,14 +22,22 @@ import (
. "github.com/apache/iggy/foreign/go/contracts"
)
-func (tms *IggyTcpClient) CreatePartition(request CreatePartitionsRequest) error {
- message := binaryserialization.CreatePartitions(request)
+func (tms *IggyTcpClient) CreatePartitions(streamId Identifier, topicId Identifier, partitionsCount uint32) error {
+ message := binaryserialization.CreatePartitions(CreatePartitionsRequest{
+ StreamId: streamId,
+ TopicId: topicId,
+ PartitionsCount: partitionsCount,
+ })
_, err := tms.sendAndFetchResponse(message, CreatePartitionsCode)
return err
}
-func (tms *IggyTcpClient) DeletePartition(request DeletePartitionRequest) error {
- message := binaryserialization.DeletePartitions(request)
+func (tms *IggyTcpClient) DeletePartitions(streamId Identifier, topicId Identifier, partitionsCount uint32) error {
+ message := binaryserialization.DeletePartitions(DeletePartitionsRequest{
+ StreamId: streamId,
+ TopicId: topicId,
+ PartitionsCount: partitionsCount,
+ })
_, err := tms.sendAndFetchResponse(message, DeletePartitionsCode)
return err
}
diff --git a/foreign/go/tcp/tcp_session_managament.go b/foreign/go/tcp/tcp_session_managament.go
index 20187ffd..2b1f75f8 100644
--- a/foreign/go/tcp/tcp_session_managament.go
+++ b/foreign/go/tcp/tcp_session_managament.go
@@ -23,8 +23,11 @@ import (
. "github.com/apache/iggy/foreign/go/contracts"
)
-func (tms *IggyTcpClient) LogIn(request LogInRequest) (*LogInResponse, error) {
- serializedRequest := binaryserialization.TcpLogInRequest{LogInRequest: request}
+func (tms *IggyTcpClient) LoginUser(username string, password string) (*IdentityInfo, error) {
+ serializedRequest := binaryserialization.TcpLogInRequest{
+ Username: username,
+ Password: password,
+ }
buffer, err := tms.sendAndFetchResponse(serializedRequest.Serialize(), LoginUserCode)
if err != nil {
return nil, err
@@ -33,8 +36,10 @@ func (tms *IggyTcpClient) LogIn(request LogInRequest) (*LogInResponse, error) {
return binaryserialization.DeserializeLogInResponse(buffer), nil
}
-func (tms *IggyTcpClient) LogInWithAccessToken(request LogInAccessTokenRequest) (*LogInResponse, error) {
- message := binaryserialization.SerializeLoginWithPersonalAccessToken(request)
+func (tms *IggyTcpClient) LoginWithPersonalAccessToken(token string) (*IdentityInfo, error) {
+ message := binaryserialization.SerializeLoginWithPersonalAccessToken(LoginWithPersonalAccessTokenRequest{
+ Token: token,
+ })
buffer, err := tms.sendAndFetchResponse(message, LoginWithAccessTokenCode)
if err != nil {
return nil, err
@@ -43,7 +48,7 @@ func (tms *IggyTcpClient) LogInWithAccessToken(request LogInAccessTokenRequest)
return binaryserialization.DeserializeLogInResponse(buffer), nil
}
-func (tms *IggyTcpClient) LogOut() error {
+func (tms *IggyTcpClient) LogoutUser() error {
_, err := tms.sendAndFetchResponse([]byte{}, LogoutUserCode)
return err
}
diff --git a/foreign/go/tcp/tcp_stream_managament.go b/foreign/go/tcp/tcp_stream_managament.go
index 8df01c26..d7a9390d 100644
--- a/foreign/go/tcp/tcp_stream_managament.go
+++ b/foreign/go/tcp/tcp_stream_managament.go
@@ -23,7 +23,7 @@ import (
ierror "github.com/apache/iggy/foreign/go/errors"
)
-func (tms *IggyTcpClient) GetStreams() ([]StreamResponse, error) {
+func (tms *IggyTcpClient) GetStreams() ([]Stream, error) {
buffer, err := tms.sendAndFetchResponse([]byte{}, GetStreamsCode)
if err != nil {
return nil, err
@@ -32,8 +32,8 @@ func (tms *IggyTcpClient) GetStreams() ([]StreamResponse, error) {
return binaryserialization.DeserializeStreams(buffer), nil
}
-func (tms *IggyTcpClient) GetStreamById(request GetStreamRequest) (*StreamResponse, error) {
- message := binaryserialization.SerializeIdentifier(request.StreamID)
+func (tms *IggyTcpClient) GetStream(streamId Identifier) (*StreamDetails, error) {
+ message := binaryserialization.SerializeIdentifier(streamId)
buffer, err := tms.sendAndFetchResponse(message, GetStreamCode)
if err != nil {
return nil, err
@@ -42,24 +42,29 @@ func (tms *IggyTcpClient) GetStreamById(request GetStreamRequest) (*StreamRespon
return nil, ierror.StreamIdNotFound
}
- stream, _ := binaryserialization.DeserializeToStream(buffer, 0)
- return &stream, nil
+ stream := binaryserialization.DeserializeStream(buffer)
+ return stream, nil
}
-func (tms *IggyTcpClient) CreateStream(request CreateStreamRequest) error {
- if MaxStringLength < len(request.Name) {
- return ierror.TextTooLong("stream_name")
+func (tms *IggyTcpClient) CreateStream(name string, streamId *uint32) (*StreamDetails, error) {
+ if MaxStringLength < len(name) {
+ return nil, ierror.TextTooLong("stream_name")
}
- serializedRequest := binaryserialization.TcpCreateStreamRequest{CreateStreamRequest: request}
- _, err := tms.sendAndFetchResponse(serializedRequest.Serialize(), CreateStreamCode)
- return err
+ serializedRequest := binaryserialization.TcpCreateStreamRequest{Name: name, StreamId: streamId}
+ buffer, err := tms.sendAndFetchResponse(serializedRequest.Serialize(), CreateStreamCode)
+ if err != nil {
+ return nil, err
+ }
+ stream := binaryserialization.DeserializeStream(buffer)
+
+ return stream, err
}
-func (tms *IggyTcpClient) UpdateStream(request UpdateStreamRequest) error {
- if MaxStringLength <= len(request.Name) {
+func (tms *IggyTcpClient) UpdateStream(streamId Identifier, name string) error {
+ if MaxStringLength <= len(name) {
return ierror.TextTooLong("stream_name")
}
- serializedRequest := binaryserialization.TcpUpdateStreamRequest{UpdateStreamRequest: request}
+ serializedRequest := binaryserialization.TcpUpdateStreamRequest{StreamId: streamId, Name: name}
_, err := tms.sendAndFetchResponse(serializedRequest.Serialize(), UpdateStreamCode)
return err
}
diff --git a/foreign/go/tcp/tcp_topic_managament.go b/foreign/go/tcp/tcp_topic_managament.go
index 0bbb7d65..64ab8c54 100644
--- a/foreign/go/tcp/tcp_topic_managament.go
+++ b/foreign/go/tcp/tcp_topic_managament.go
@@ -18,12 +18,14 @@
package tcp
import (
+ "time"
+
binaryserialization "github.com/apache/iggy/foreign/go/binary_serialization"
. "github.com/apache/iggy/foreign/go/contracts"
ierror "github.com/apache/iggy/foreign/go/errors"
)
-func (tms *IggyTcpClient) GetTopics(streamId Identifier) ([]TopicResponse, error) {
+func (tms *IggyTcpClient) GetTopics(streamId Identifier) ([]Topic, error) {
message := binaryserialization.SerializeIdentifier(streamId)
buffer, err := tms.sendAndFetchResponse(message, GetTopicsCode)
if err != nil {
@@ -33,7 +35,7 @@ func (tms *IggyTcpClient) GetTopics(streamId Identifier) ([]TopicResponse, error
return binaryserialization.DeserializeTopics(buffer)
}
-func (tms *IggyTcpClient) GetTopicById(streamId Identifier, topicId Identifier) (*TopicResponse, error) {
+func (tms *IggyTcpClient) GetTopic(streamId Identifier, topicId Identifier) (*TopicDetails, error) {
message := binaryserialization.SerializeIdentifiers(streamId, topicId)
buffer, err := tms.sendAndFetchResponse(message, GetTopicCode)
if err != nil {
@@ -51,20 +53,57 @@ func (tms *IggyTcpClient) GetTopicById(streamId Identifier, topicId Identifier)
return topic, nil
}
-func (tms *IggyTcpClient) CreateTopic(request CreateTopicRequest) error {
- if MaxStringLength < len(request.Name) {
- return ierror.TextTooLong("topic_name")
+func (tms *IggyTcpClient) CreateTopic(
+ streamId Identifier,
+ name string,
+ partitionsCount int,
+ compressionAlgorithm uint8,
+ messageExpiry time.Duration,
+ maxTopicSize uint64,
+ replicationFactor *uint8,
+ topicId *int,
+) (*TopicDetails, error) {
+ if MaxStringLength < len(name) {
+ return nil, ierror.TextTooLong("topic_name")
}
- serializedRequest := binaryserialization.TcpCreateTopicRequest{CreateTopicRequest: request}
- _, err := tms.sendAndFetchResponse(serializedRequest.Serialize(), CreateTopicCode)
- return err
+ serializedRequest := binaryserialization.TcpCreateTopicRequest{
+ StreamId: streamId,
+ Name: name,
+ PartitionsCount: partitionsCount,
+ CompressionAlgorithm: compressionAlgorithm,
+ MessageExpiry: messageExpiry,
+ MaxTopicSize: maxTopicSize,
+ ReplicationFactor: replicationFactor,
+ TopicId: topicId,
+ }
+ buffer, err := tms.sendAndFetchResponse(serializedRequest.Serialize(), CreateTopicCode)
+ if err != nil {
+ return nil, err
+ }
+ topic, err := binaryserialization.DeserializeTopic(buffer)
+ return topic, err
}
-func (tms *IggyTcpClient) UpdateTopic(request UpdateTopicRequest) error {
- if MaxStringLength < len(request.Name) {
+func (tms *IggyTcpClient) UpdateTopic(
+ streamId Identifier,
+ topicId Identifier,
+ name string,
+ compressionAlgorithm uint8,
+ messageExpiry time.Duration,
+ maxTopicSize uint64,
+ replicationFactor *uint8,
+) error {
+ if MaxStringLength < len(name) {
return ierror.TextTooLong("topic_name")
}
- serializedRequest := binaryserialization.TcpUpdateTopicRequest{UpdateTopicRequest: request}
+ serializedRequest := binaryserialization.TcpUpdateTopicRequest{
+ StreamId: streamId,
+ TopicId: topicId,
+ CompressionAlgorithm: compressionAlgorithm,
+ MessageExpiry: messageExpiry,
+ MaxTopicSize: maxTopicSize,
+ ReplicationFactor: replicationFactor,
+ Name: name}
_, err := tms.sendAndFetchResponse(serializedRequest.Serialize(), UpdateTopicCode)
return err
}
diff --git a/foreign/go/tcp/tcp_user_managament.go b/foreign/go/tcp/tcp_user_managament.go
index 26bf4d98..736249af 100644
--- a/foreign/go/tcp/tcp_user_managament.go
+++ b/foreign/go/tcp/tcp_user_managament.go
@@ -23,7 +23,7 @@ import (
ierror "github.com/apache/iggy/foreign/go/errors"
)
-func (tms *IggyTcpClient) GetUser(identifier Identifier) (*UserResponse, error) {
+func (tms *IggyTcpClient) GetUser(identifier Identifier) (*UserInfoDetails, error) {
message := binaryserialization.SerializeIdentifier(identifier)
buffer, err := tms.sendAndFetchResponse(message, GetUserCode)
if err != nil {
@@ -36,7 +36,7 @@ func (tms *IggyTcpClient) GetUser(identifier Identifier) (*UserResponse, error)
return binaryserialization.DeserializeUser(buffer)
}
-func (tms *IggyTcpClient) GetUsers() ([]*UserResponse, error) {
+func (tms *IggyTcpClient) GetUsers() ([]UserInfo, error) {
buffer, err := tms.sendAndFetchResponse([]byte{}, GetUsersCode)
if err != nil {
return nil, err
@@ -45,14 +45,30 @@ func (tms *IggyTcpClient) GetUsers() ([]*UserResponse, error) {
return binaryserialization.DeserializeUsers(buffer)
}
-func (tms *IggyTcpClient) CreateUser(request CreateUserRequest) error {
- message := binaryserialization.SerializeCreateUserRequest(request)
- _, err := tms.sendAndFetchResponse(message, CreateUserCode)
- return err
+func (tms *IggyTcpClient) CreateUser(username string, password string, status UserStatus, permissions *Permissions) (*UserInfoDetails, error) {
+ message := binaryserialization.SerializeCreateUserRequest(CreateUserRequest{
+ Username: username,
+ Password: password,
+ Status: status,
+ Permissions: permissions,
+ })
+ buffer, err := tms.sendAndFetchResponse(message, CreateUserCode)
+ if err != nil {
+ return nil, err
+ }
+ userInfo, err := binaryserialization.DeserializeUser(buffer)
+ if err != nil {
+ return nil, err
+ }
+ return userInfo, nil
}
-func (tms *IggyTcpClient) UpdateUser(request UpdateUserRequest) error {
- message := binaryserialization.SerializeUpdateUser(request)
+func (tms *IggyTcpClient) UpdateUser(userID Identifier, username *string, status *UserStatus) error {
+ message := binaryserialization.SerializeUpdateUser(UpdateUserRequest{
+ UserID: userID,
+ Username: username,
+ Status: status,
+ })
_, err := tms.sendAndFetchResponse(message, UpdateUserCode)
return err
}
@@ -63,14 +79,21 @@ func (tms *IggyTcpClient) DeleteUser(identifier Identifier) error {
return err
}
-func (tms *IggyTcpClient) UpdateUserPermissions(request UpdateUserPermissionsRequest) error {
- message := binaryserialization.SerializeUpdateUserPermissionsRequest(request)
+func (tms *IggyTcpClient) UpdatePermissions(userID Identifier, permissions *Permissions) error {
+ message := binaryserialization.SerializeUpdateUserPermissionsRequest(UpdatePermissionsRequest{
+ UserID: userID,
+ Permissions: permissions,
+ })
_, err := tms.sendAndFetchResponse(message, UpdatePermissionsCode)
return err
}
-func (tms *IggyTcpClient) ChangePassword(request ChangePasswordRequest) error {
- message := binaryserialization.SerializeChangePasswordRequest(request)
+func (tms *IggyTcpClient) ChangePassword(userID Identifier, currentPassword string, newPassword string) error {
+ message := binaryserialization.SerializeChangePasswordRequest(ChangePasswordRequest{
+ UserID: userID,
+ CurrentPassword: currentPassword,
+ NewPassword: newPassword,
+ })
_, err := tms.sendAndFetchResponse(message, ChangePasswordCode)
return err
}