diff --git a/docker-compose.yml b/docker-compose.yml
index 8f1eb50b2e1c42b2262e02a25755841772888d36..47346983c8ab2341cbb2b403477fdad386d96100 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -3,6 +3,8 @@ services:
build:
context: ./src
dockerfile: banktransfer/Dockerfile
+ environment:
+ - KAFKA_CONNECT=kafka:9092
myaktion:
build:
context: ./src
@@ -17,4 +19,19 @@ services:
image: mariadb:10.5
environment:
- MYSQL_ROOT_PASSWORD=root
- - MYSQL_DATABASE=myaktion
\ No newline at end of file
+ - MYSQL_DATABASE=myaktion
+ kafka:
+ image: bitnami/kafka:3.3.2
+ container_name: kafka
+ environment:
+ - KAFKA_ENABLE_KRAFT=yes
+ - KAFKA_CFG_NODE_ID=1
+ - KAFKA_CFG_PROCESS_ROLES=broker,controller
+ - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
+ - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093
+ - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
+ - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092
+ - KAFKA_CFG_BROKER_ID=1
+ - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@kafka:9093
+ - ALLOW_PLAINTEXT_LISTENER=yes
+ - KAFKA_KRAFT_CLUSTER_ID=r4zt_wrqTRuT7W2NJsB_GA
\ No newline at end of file
diff --git a/src/banktransfer/Dockerfile b/src/banktransfer/Dockerfile
index 808cebe062459b3c434fd713d79b76e226d8cf4d..652a65cf5b250634edfec389f1adacbe3e3ce359 100644
--- a/src/banktransfer/Dockerfile
+++ b/src/banktransfer/Dockerfile
@@ -13,6 +13,10 @@ RUN go mod download
RUN go generate ./...
RUN go install
+RUN wget https://raw.githubusercontent.com/vishnubob/wait-for-it/81b1373f17855a4dc21156cfe1694c31d7d1792e/wait-for-it.sh
+RUN chmod +x ./wait-for-it.sh ./docker-entrypoint.sh
+
+ENTRYPOINT ["./docker-entrypoint.sh"]
CMD ["banktransfer"]
EXPOSE 9111
diff --git a/src/banktransfer/docker-entrypoint.sh b/src/banktransfer/docker-entrypoint.sh
new file mode 100644
index 0000000000000000000000000000000000000000..f86a6699841828cbe552075320c7e2a390956c54
--- /dev/null
+++ b/src/banktransfer/docker-entrypoint.sh
@@ -0,0 +1,14 @@
+#!/bin/sh
+
+# Abort on any error (including if wait-for-it fails).
+set -e
+
+# Wait for kafka
+if [ -n "$KAFKA_CONNECT" ]; then
+ /go/src/app/wait-for-it.sh "$KAFKA_CONNECT" -t 120
+ # we need to wait a bit more, because sometimes kafka is not yet ready
+ sleep 1
+fi
+
+# Run the main container command.
+exec "$@"
\ No newline at end of file
diff --git a/src/banktransfer/go.mod b/src/banktransfer/go.mod
index ee8fafd6ca98e4e4d1ca36e58549fd21dd4328cd..45d81b9b38ff3be877281d7855aafa184b64e147 100644
--- a/src/banktransfer/go.mod
+++ b/src/banktransfer/go.mod
@@ -3,13 +3,16 @@ module gitlab.reutlingen-university.de/albrecht/myaktion-go/src/banktransfer
go 1.20
require (
+ github.com/golang/protobuf v1.5.3
+ github.com/segmentio/kafka-go v0.4.40
github.com/sirupsen/logrus v1.9.0
google.golang.org/grpc v1.55.0
google.golang.org/protobuf v1.30.0
)
require (
- github.com/golang/protobuf v1.5.3 // indirect
+ github.com/klauspost/compress v1.15.9 // indirect
+ github.com/pierrec/lz4/v4 v4.1.15 // indirect
golang.org/x/net v0.8.0 // indirect
golang.org/x/sys v0.6.0 // indirect
golang.org/x/text v0.8.0 // indirect
diff --git a/src/banktransfer/go.sum b/src/banktransfer/go.sum
index fb01c1e07192678cd415679aee90e5d3ff41b917..ac6641ab98494a29fa41f0b023570ecc1ee0feef 100644
--- a/src/banktransfer/go.sum
+++ b/src/banktransfer/go.sum
@@ -6,20 +6,59 @@ github.com/golang/protobuf v1.5.3 h1:KhyjKVUg7Usr/dYsdSqoFveMYd5ko72D+zANwlG1mmg
github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
+github.com/klauspost/compress v1.15.9 h1:wKRjX6JRtDdrE9qwa4b/Cip7ACOshUI4smpCQanqjSY=
+github.com/klauspost/compress v1.15.9/go.mod h1:PhcZ0MbTNciWF3rruxRgKxI5NkcHHrHUDtV4Yw2GlzU=
+github.com/pierrec/lz4/v4 v4.1.15 h1:MO0/ucJhngq7299dKLwIMtgTfbkoSPF6AoMYDd8Q4q0=
+github.com/pierrec/lz4/v4 v4.1.15/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/segmentio/kafka-go v0.4.40 h1:sszW7c0/uyv7+VcTW5trx2ZC7kMWDTxuR/6Zn8U1bm8=
+github.com/segmentio/kafka-go v0.4.40/go.mod h1:naFEZc5MQKdeL3W6NkZIAn48Y6AazqjRFDhnXeg3h94=
github.com/sirupsen/logrus v1.9.0 h1:trlNQbNUG3OdDrDil03MCb1H2o9nJ1x4/5LYw7byDE0=
github.com/sirupsen/logrus v1.9.0/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
-github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
+github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.8.0 h1:pSgiaMZlXftHpm5L7V1+rVB+AZJydKsMxsQBIJw4PKk=
+github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
+github.com/xdg-go/pbkdf2 v1.0.0 h1:Su7DPu48wXMwC3bs7MCNG+z4FhcyEuz5dlvchbq0B0c=
+github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI=
+github.com/xdg-go/scram v1.1.2 h1:FHX5I5B4i4hKRVRBCFRxq1iQRej7WO3hhBuJf+UUySY=
+github.com/xdg-go/scram v1.1.2/go.mod h1:RT/sEzTbU5y00aCK8UOx6R7YryM0iF1N2MOmC3kKLN4=
+github.com/xdg-go/stringprep v1.0.4 h1:XLI/Ng3O1Atzq0oBs3TWm+5ZVgkq2aqdlvP9JtoZ6c8=
+github.com/xdg-go/stringprep v1.0.4/go.mod h1:mPGuuIYwz7CmR2bT9j4GbQqutWS1zV24gijq1dTyGkM=
+github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
+golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
+golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
+golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
+golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
golang.org/x/net v0.8.0 h1:Zrh2ngAOFYneWTAIAPethzeaQLuHwhuBkuV6ZiRnUaQ=
golang.org/x/net v0.8.0/go.mod h1:QVkue5JL9kW//ek3r6jTKnTFis1tRmNAW2P1shuFdJc=
+golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.6.0 h1:MVltZSvRTcU2ljQOhs94SXPftV6DCNnZViHeQps87pQ=
golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
+golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
+golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
+golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ=
golang.org/x/text v0.8.0 h1:57P1ETyNKtuIjB4SRd15iJxuhj8Gc416Y78H3qgMh68=
golang.org/x/text v0.8.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
+golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
+golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
+golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
google.golang.org/genproto v0.0.0-20230306155012-7f2fa6fef1f4 h1:DdoeryqhaXp1LtT/emMP1BRJPHHKFi5akj/nbx/zNTA=
google.golang.org/genproto v0.0.0-20230306155012-7f2fa6fef1f4/go.mod h1:NWraEVixdDnqcqQ30jipen1STv2r/n24Wb7twVTGR4s=
@@ -30,5 +69,6 @@ google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQ
google.golang.org/protobuf v1.30.0 h1:kPPoIgf3TsEvrm0PFe15JQ+570QVxYzEvvHqChK+cng=
google.golang.org/protobuf v1.30.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
diff --git a/src/banktransfer/grpc/banktransfer/banktransfer.pb.go b/src/banktransfer/grpc/banktransfer/banktransfer.pb.go
index 93f5ffc07e107e63fba5d273c5a69c674dad5d0e..deb7e5818ca69c76a433e4611f5822ac76c31c6d 100644
--- a/src/banktransfer/grpc/banktransfer/banktransfer.pb.go
+++ b/src/banktransfer/grpc/banktransfer/banktransfer.pb.go
@@ -1,15 +1,15 @@
// Code generated by protoc-gen-go. DO NOT EDIT.
// versions:
-// protoc-gen-go v1.26.0
-// protoc v3.17.0
+// protoc-gen-go v1.30.0
+// protoc v3.12.4
// source: banktransfer/banktransfer.proto
package banktransfer
import (
+ empty "github.com/golang/protobuf/ptypes/empty"
protoreflect "google.golang.org/protobuf/reflect/protoreflect"
protoimpl "google.golang.org/protobuf/runtime/protoimpl"
- emptypb "google.golang.org/protobuf/types/known/emptypb"
reflect "reflect"
sync "sync"
)
@@ -89,7 +89,7 @@ type Transaction struct {
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
- Id int32 `protobuf:"varint,1,opt,name=id,proto3" json:"id,omitempty"`
+ Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`
DonationId int32 `protobuf:"varint,2,opt,name=donation_id,json=donationId,proto3" json:"donation_id,omitempty"`
Amount float32 `protobuf:"fixed32,3,opt,name=amount,proto3" json:"amount,omitempty"`
Reference string `protobuf:"bytes,4,opt,name=reference,proto3" json:"reference,omitempty"`
@@ -129,11 +129,11 @@ func (*Transaction) Descriptor() ([]byte, []int) {
return file_banktransfer_banktransfer_proto_rawDescGZIP(), []int{1}
}
-func (x *Transaction) GetId() int32 {
+func (x *Transaction) GetId() string {
if x != nil {
return x.Id
}
- return 0
+ return ""
}
func (x *Transaction) GetDonationId() int32 {
@@ -176,7 +176,7 @@ type ProcessingResponse struct {
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
- Id int32 `protobuf:"varint,1,opt,name=id,proto3" json:"id,omitempty"`
+ Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`
}
func (x *ProcessingResponse) Reset() {
@@ -211,11 +211,11 @@ func (*ProcessingResponse) Descriptor() ([]byte, []int) {
return file_banktransfer_banktransfer_proto_rawDescGZIP(), []int{2}
}
-func (x *ProcessingResponse) GetId() int32 {
+func (x *ProcessingResponse) GetId() string {
if x != nil {
return x.Id
}
- return 0
+ return ""
}
var File_banktransfer_banktransfer_proto protoreflect.FileDescriptor
@@ -232,7 +232,7 @@ var file_banktransfer_banktransfer_proto_rawDesc = []byte{
0x62, 0x61, 0x6e, 0x6b, 0x4e, 0x61, 0x6d, 0x65, 0x12, 0x16, 0x0a, 0x06, 0x6e, 0x75, 0x6d, 0x62,
0x65, 0x72, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x6e, 0x75, 0x6d, 0x62, 0x65, 0x72,
0x22, 0xe4, 0x01, 0x0a, 0x0b, 0x54, 0x72, 0x61, 0x6e, 0x73, 0x61, 0x63, 0x74, 0x69, 0x6f, 0x6e,
- 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x05, 0x52, 0x02, 0x69, 0x64,
+ 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x02, 0x69, 0x64,
0x12, 0x1f, 0x0a, 0x0b, 0x64, 0x6f, 0x6e, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x69, 0x64, 0x18,
0x02, 0x20, 0x01, 0x28, 0x05, 0x52, 0x0a, 0x64, 0x6f, 0x6e, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x49,
0x64, 0x12, 0x16, 0x0a, 0x06, 0x61, 0x6d, 0x6f, 0x75, 0x6e, 0x74, 0x18, 0x03, 0x20, 0x01, 0x28,
@@ -247,7 +247,7 @@ var file_banktransfer_banktransfer_proto_rawDesc = []byte{
0x73, 0x66, 0x65, 0x72, 0x2e, 0x41, 0x63, 0x63, 0x6f, 0x75, 0x6e, 0x74, 0x52, 0x09, 0x74, 0x6f,
0x41, 0x63, 0x63, 0x6f, 0x75, 0x6e, 0x74, 0x22, 0x24, 0x0a, 0x12, 0x50, 0x72, 0x6f, 0x63, 0x65,
0x73, 0x73, 0x69, 0x6e, 0x67, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x0e, 0x0a,
- 0x02, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x05, 0x52, 0x02, 0x69, 0x64, 0x32, 0xae, 0x01,
+ 0x02, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x02, 0x69, 0x64, 0x32, 0xae, 0x01,
0x0a, 0x0c, 0x42, 0x61, 0x6e, 0x6b, 0x54, 0x72, 0x61, 0x6e, 0x73, 0x66, 0x65, 0x72, 0x12, 0x44,
0x0a, 0x0d, 0x54, 0x72, 0x61, 0x6e, 0x73, 0x66, 0x65, 0x72, 0x4d, 0x6f, 0x6e, 0x65, 0x79, 0x12,
0x19, 0x2e, 0x62, 0x61, 0x6e, 0x6b, 0x74, 0x72, 0x61, 0x6e, 0x73, 0x66, 0x65, 0x72, 0x2e, 0x54,
@@ -258,14 +258,14 @@ var file_banktransfer_banktransfer_proto_rawDesc = []byte{
0x6e, 0x6b, 0x74, 0x72, 0x61, 0x6e, 0x73, 0x66, 0x65, 0x72, 0x2e, 0x50, 0x72, 0x6f, 0x63, 0x65,
0x73, 0x73, 0x69, 0x6e, 0x67, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x1a, 0x19, 0x2e,
0x62, 0x61, 0x6e, 0x6b, 0x74, 0x72, 0x61, 0x6e, 0x73, 0x66, 0x65, 0x72, 0x2e, 0x54, 0x72, 0x61,
- 0x6e, 0x73, 0x61, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x22, 0x00, 0x28, 0x01, 0x30, 0x01, 0x42, 0x64,
- 0x5a, 0x62, 0x67, 0x69, 0x74, 0x6c, 0x61, 0x62, 0x2e, 0x72, 0x65, 0x75, 0x74, 0x6c, 0x69, 0x6e,
+ 0x6e, 0x73, 0x61, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x22, 0x00, 0x28, 0x01, 0x30, 0x01, 0x42, 0x59,
+ 0x5a, 0x57, 0x67, 0x69, 0x74, 0x6c, 0x61, 0x62, 0x2e, 0x72, 0x65, 0x75, 0x74, 0x6c, 0x69, 0x6e,
0x67, 0x65, 0x6e, 0x2d, 0x75, 0x6e, 0x69, 0x76, 0x65, 0x72, 0x73, 0x69, 0x74, 0x79, 0x2e, 0x64,
- 0x65, 0x2f, 0x67, 0x6f, 0x2d, 0x65, 0x78, 0x65, 0x72, 0x63, 0x69, 0x73, 0x65, 0x73, 0x2f, 0x6d,
- 0x79, 0x61, 0x6b, 0x74, 0x69, 0x6f, 0x6e, 0x2d, 0x67, 0x6f, 0x2d, 0x73, 0x73, 0x32, 0x30, 0x32,
- 0x33, 0x2f, 0x73, 0x72, 0x63, 0x2f, 0x62, 0x61, 0x6e, 0x6b, 0x74, 0x72, 0x61, 0x6e, 0x73, 0x66,
- 0x65, 0x72, 0x2f, 0x67, 0x72, 0x70, 0x63, 0x2f, 0x62, 0x61, 0x6e, 0x6b, 0x74, 0x72, 0x61, 0x6e,
- 0x73, 0x66, 0x65, 0x72, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+ 0x65, 0x2f, 0x61, 0x6c, 0x62, 0x72, 0x65, 0x63, 0x68, 0x74, 0x2f, 0x6d, 0x79, 0x61, 0x6b, 0x74,
+ 0x69, 0x6f, 0x6e, 0x2d, 0x67, 0x6f, 0x2f, 0x73, 0x72, 0x63, 0x2f, 0x62, 0x61, 0x6e, 0x6b, 0x74,
+ 0x72, 0x61, 0x6e, 0x73, 0x66, 0x65, 0x72, 0x2f, 0x67, 0x72, 0x70, 0x63, 0x2f, 0x62, 0x61, 0x6e,
+ 0x6b, 0x74, 0x72, 0x61, 0x6e, 0x73, 0x66, 0x65, 0x72, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f,
+ 0x33,
}
var (
@@ -285,7 +285,7 @@ var file_banktransfer_banktransfer_proto_goTypes = []interface{}{
(*Account)(nil), // 0: banktransfer.Account
(*Transaction)(nil), // 1: banktransfer.Transaction
(*ProcessingResponse)(nil), // 2: banktransfer.ProcessingResponse
- (*emptypb.Empty)(nil), // 3: google.protobuf.Empty
+ (*empty.Empty)(nil), // 3: google.protobuf.Empty
}
var file_banktransfer_banktransfer_proto_depIdxs = []int32{
0, // 0: banktransfer.Transaction.from_account:type_name -> banktransfer.Account
diff --git a/src/banktransfer/grpc/banktransfer/banktransfer.proto b/src/banktransfer/grpc/banktransfer/banktransfer.proto
index c7de1c3c8a9a84cc57a1b69aa73baeb6687d7f1d..db3aa7f067c6f73d0b6fa9644923ec4ef05dd3a1 100644
--- a/src/banktransfer/grpc/banktransfer/banktransfer.proto
+++ b/src/banktransfer/grpc/banktransfer/banktransfer.proto
@@ -18,7 +18,7 @@ message Account {
}
message Transaction {
- int32 id = 1;
+ string id = 1;
int32 donation_id = 2;
float amount = 3;
string reference = 4;
@@ -27,5 +27,5 @@ message Transaction {
}
message ProcessingResponse {
- int32 id = 1;
+ string id = 1;
}
diff --git a/src/banktransfer/grpc/banktransfer/banktransfer_grpc.pb.go b/src/banktransfer/grpc/banktransfer/banktransfer_grpc.pb.go
index 02af0f7c343852c91d51beb9b2850b4e8be423fb..edc2f69bb4c4cb43109851e8c35fe280e995c79a 100644
--- a/src/banktransfer/grpc/banktransfer/banktransfer_grpc.pb.go
+++ b/src/banktransfer/grpc/banktransfer/banktransfer_grpc.pb.go
@@ -1,13 +1,17 @@
// Code generated by protoc-gen-go-grpc. DO NOT EDIT.
+// versions:
+// - protoc-gen-go-grpc v1.3.0
+// - protoc v3.12.4
+// source: banktransfer/banktransfer.proto
package banktransfer
import (
context "context"
+ empty "github.com/golang/protobuf/ptypes/empty"
grpc "google.golang.org/grpc"
codes "google.golang.org/grpc/codes"
status "google.golang.org/grpc/status"
- emptypb "google.golang.org/protobuf/types/known/emptypb"
)
// This is a compile-time assertion to ensure that this generated file
@@ -15,11 +19,16 @@ import (
// Requires gRPC-Go v1.32.0 or later.
const _ = grpc.SupportPackageIsVersion7
+const (
+ BankTransfer_TransferMoney_FullMethodName = "/banktransfer.BankTransfer/TransferMoney"
+ BankTransfer_ProcessTransactions_FullMethodName = "/banktransfer.BankTransfer/ProcessTransactions"
+)
+
// BankTransferClient is the client API for BankTransfer service.
//
// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.
type BankTransferClient interface {
- TransferMoney(ctx context.Context, in *Transaction, opts ...grpc.CallOption) (*emptypb.Empty, error)
+ TransferMoney(ctx context.Context, in *Transaction, opts ...grpc.CallOption) (*empty.Empty, error)
ProcessTransactions(ctx context.Context, opts ...grpc.CallOption) (BankTransfer_ProcessTransactionsClient, error)
}
@@ -31,9 +40,9 @@ func NewBankTransferClient(cc grpc.ClientConnInterface) BankTransferClient {
return &bankTransferClient{cc}
}
-func (c *bankTransferClient) TransferMoney(ctx context.Context, in *Transaction, opts ...grpc.CallOption) (*emptypb.Empty, error) {
- out := new(emptypb.Empty)
- err := c.cc.Invoke(ctx, "/banktransfer.BankTransfer/TransferMoney", in, out, opts...)
+func (c *bankTransferClient) TransferMoney(ctx context.Context, in *Transaction, opts ...grpc.CallOption) (*empty.Empty, error) {
+ out := new(empty.Empty)
+ err := c.cc.Invoke(ctx, BankTransfer_TransferMoney_FullMethodName, in, out, opts...)
if err != nil {
return nil, err
}
@@ -41,7 +50,7 @@ func (c *bankTransferClient) TransferMoney(ctx context.Context, in *Transaction,
}
func (c *bankTransferClient) ProcessTransactions(ctx context.Context, opts ...grpc.CallOption) (BankTransfer_ProcessTransactionsClient, error) {
- stream, err := c.cc.NewStream(ctx, &BankTransfer_ServiceDesc.Streams[0], "/banktransfer.BankTransfer/ProcessTransactions", opts...)
+ stream, err := c.cc.NewStream(ctx, &BankTransfer_ServiceDesc.Streams[0], BankTransfer_ProcessTransactions_FullMethodName, opts...)
if err != nil {
return nil, err
}
@@ -75,7 +84,7 @@ func (x *bankTransferProcessTransactionsClient) Recv() (*Transaction, error) {
// All implementations must embed UnimplementedBankTransferServer
// for forward compatibility
type BankTransferServer interface {
- TransferMoney(context.Context, *Transaction) (*emptypb.Empty, error)
+ TransferMoney(context.Context, *Transaction) (*empty.Empty, error)
ProcessTransactions(BankTransfer_ProcessTransactionsServer) error
mustEmbedUnimplementedBankTransferServer()
}
@@ -84,7 +93,7 @@ type BankTransferServer interface {
type UnimplementedBankTransferServer struct {
}
-func (UnimplementedBankTransferServer) TransferMoney(context.Context, *Transaction) (*emptypb.Empty, error) {
+func (UnimplementedBankTransferServer) TransferMoney(context.Context, *Transaction) (*empty.Empty, error) {
return nil, status.Errorf(codes.Unimplemented, "method TransferMoney not implemented")
}
func (UnimplementedBankTransferServer) ProcessTransactions(BankTransfer_ProcessTransactionsServer) error {
@@ -113,7 +122,7 @@ func _BankTransfer_TransferMoney_Handler(srv interface{}, ctx context.Context, d
}
info := &grpc.UnaryServerInfo{
Server: srv,
- FullMethod: "/banktransfer.BankTransfer/TransferMoney",
+ FullMethod: BankTransfer_TransferMoney_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(BankTransferServer).TransferMoney(ctx, req.(*Transaction))
diff --git a/src/banktransfer/kafka/reader.go b/src/banktransfer/kafka/reader.go
new file mode 100644
index 0000000000000000000000000000000000000000..8a7d4b70bd6650546c941899ca8ffdb2063ce232
--- /dev/null
+++ b/src/banktransfer/kafka/reader.go
@@ -0,0 +1,60 @@
+package kafka
+
+import (
+ "context"
+ "encoding/json"
+ "fmt"
+
+ "github.com/segmentio/kafka-go"
+ log "github.com/sirupsen/logrus"
+ "gitlab.reutlingen-university.de/albrecht/myaktion-go/src/banktransfer/grpc/banktransfer"
+)
+
+type TransactionReader interface {
+ Close() error
+ Read(ctx context.Context, handler func(*banktransfer.Transaction) error) error
+}
+
+type ReadHandler func(*banktransfer.Transaction) error
+
+type reader struct {
+ reader *kafka.Reader
+}
+
+func NewTransactionReader() *reader {
+ return &reader{reader: kafka.NewReader(kafka.ReaderConfig{
+ Brokers: []string{connect},
+ GroupID: groupID,
+ Topic: Topic,
+ MinBytes: 10e2, // 1KB
+ MaxBytes: 10e5, // 1MB
+ })}
+}
+
+func (r *reader) Read(ctx context.Context, handler ReadHandler) error {
+ msg, err := r.reader.FetchMessage(ctx)
+ if err != nil {
+ return fmt.Errorf("failed to fetch message from Kafka server: %w", err)
+ }
+ entry := log.WithField("message", msg)
+ entry.Trace("received message from Kafka server")
+ var transaction banktransfer.Transaction
+ if err := json.Unmarshal(msg.Value, &transaction); err != nil {
+ return fmt.Errorf("failed to unmarshal message: %w", err)
+ }
+ err = handler(&transaction)
+ if err != nil {
+ return err
+ }
+ if err := r.reader.CommitMessages(ctx, msg); err != nil {
+ return fmt.Errorf("failed to commit message: %w", err)
+ }
+ return nil
+}
+
+func (r *reader) Close() error {
+ if err := r.reader.Close(); err != nil {
+ return fmt.Errorf("could not close Kafka subscription: %w", err)
+ }
+ return nil
+}
diff --git a/src/banktransfer/kafka/topic.go b/src/banktransfer/kafka/topic.go
new file mode 100644
index 0000000000000000000000000000000000000000..0f14f4ca57f76d1aa8200f997389b39c3fb502c1
--- /dev/null
+++ b/src/banktransfer/kafka/topic.go
@@ -0,0 +1,51 @@
+package kafka
+
+import (
+ "net"
+ "strconv"
+
+ "github.com/segmentio/kafka-go"
+ log "github.com/sirupsen/logrus"
+)
+
+const (
+ Topic = "banktransfer"
+)
+
+func EnsureTransactionTopic() {
+ if err := ensureTopic(Topic, 10); err != nil {
+ panic(err.Error())
+ }
+}
+
+func ensureTopic(topic string, numPartitions int) error {
+ conn, err := kafka.Dial("tcp", connect)
+ if err != nil {
+ log.Warnf("Tcp conn failed: %v", err)
+ }
+
+ defer conn.Close()
+ controller, err := conn.Controller()
+ if err != nil {
+ log.Errorf("error conn.Controller: %v", err)
+ }
+
+ var leaderConn *kafka.Conn
+ leaderConn, err = kafka.Dial("tcp", net.JoinHostPort(controller.Host, strconv.Itoa(controller.Port)))
+ if err != nil {
+ log.Errorf("error kafka.Dial: %v", err)
+ }
+ defer leaderConn.Close()
+ topicConfigs := []kafka.TopicConfig{
+ {
+ Topic: topic,
+ NumPartitions: numPartitions,
+ ReplicationFactor: 1,
+ },
+ }
+ err = leaderConn.CreateTopics(topicConfigs...)
+ if err != nil {
+ log.Errorf("error in CreateTopics: %v", err)
+ }
+ return nil
+}
diff --git a/src/banktransfer/kafka/writer.go b/src/banktransfer/kafka/writer.go
new file mode 100644
index 0000000000000000000000000000000000000000..4ee29af5e0665b720585ff610da5865129e6d37a
--- /dev/null
+++ b/src/banktransfer/kafka/writer.go
@@ -0,0 +1,59 @@
+package kafka
+
+import (
+ "context"
+ "encoding/json"
+ "fmt"
+ "os"
+
+ "github.com/segmentio/kafka-go"
+ "gitlab.reutlingen-university.de/albrecht/myaktion-go/src/banktransfer/grpc/banktransfer"
+)
+
+const (
+ groupID = "banktransfer-watchers"
+)
+
+var connect = os.Getenv("KAFKA_CONNECT")
+
+type TransactionWriter interface {
+ Close() error
+ Write(transaction *banktransfer.Transaction) error
+}
+type writer struct {
+ writer *kafka.Writer
+ writeContext context.Context
+ writeCancelFunc context.CancelFunc
+}
+
+func NewTransactionWriter() *writer {
+ writeContext, writeCancelFunc := context.WithCancel(context.Background())
+ return &writer{&kafka.Writer{
+ Addr: kafka.TCP(connect),
+ Topic: Topic,
+ Balancer: &kafka.LeastBytes{},
+ }, writeContext, writeCancelFunc,
+ }
+}
+
+func (w *writer) Write(transaction *banktransfer.Transaction) error {
+ bytes, err := json.Marshal(&transaction)
+ if err != nil {
+ return fmt.Errorf("can't marshal transaction to JSON: %w", err)
+ }
+ if err := w.writer.WriteMessages(w.writeContext, kafka.Message{
+ Key: []byte(transaction.Id),
+ Value: bytes,
+ }); err != nil {
+ return fmt.Errorf("can't write message to kafka server: %w", err)
+ }
+ return nil
+}
+
+func (w *writer) Close() error {
+ w.writeCancelFunc()
+ if err := w.writer.Close(); err != nil {
+ return fmt.Errorf("couldn't close connection to Kafka server: %w", err)
+ }
+ return nil
+}
diff --git a/src/banktransfer/main.go b/src/banktransfer/main.go
index 01c9753908b2ede887a45ba42c21de4089e89bfa..f150850ac2cd9d8545fc41266b0a07ab91427bd1 100644
--- a/src/banktransfer/main.go
+++ b/src/banktransfer/main.go
@@ -7,11 +7,13 @@ import (
log "github.com/sirupsen/logrus"
"gitlab.reutlingen-university.de/albrecht/myaktion-go/src/banktransfer/grpc/banktransfer"
+ "gitlab.reutlingen-university.de/albrecht/myaktion-go/src/banktransfer/kafka"
"gitlab.reutlingen-university.de/albrecht/myaktion-go/src/banktransfer/service"
"google.golang.org/grpc"
)
func init() {
+ defer kafka.EnsureTransactionTopic()
// init logger
log.SetFormatter(&log.TextFormatter{})
log.SetReportCaller(true)
diff --git a/src/banktransfer/service/banktransfer.go b/src/banktransfer/service/banktransfer.go
index 8b840cf00cccc0c8d6268c11e18e3579e490c4a0..9ac1b41a6ebdfc38405431c651df41e4f096b6b6 100644
--- a/src/banktransfer/service/banktransfer.go
+++ b/src/banktransfer/service/banktransfer.go
@@ -2,32 +2,23 @@ package service
import (
"context"
- "math/rand"
- "sync/atomic"
- "time"
+ "errors"
+ "fmt"
log "github.com/sirupsen/logrus"
"gitlab.reutlingen-university.de/albrecht/myaktion-go/src/banktransfer/grpc/banktransfer"
+ "gitlab.reutlingen-university.de/albrecht/myaktion-go/src/banktransfer/kafka"
"google.golang.org/protobuf/types/known/emptypb"
)
-const retryTime = 5 * time.Second
-
type BankTransferService struct {
banktransfer.BankTransferServer
- counter int32
- queue chan banktransfer.Transaction
- retryQueue chan banktransfer.Transaction
- stop chan struct{}
+ keyGenerator *KeyGenerator
+ transactionWriter kafka.TransactionWriter
}
func NewBankTransferService() *BankTransferService {
- rand.Seed(time.Now().UnixNano())
- return &BankTransferService{counter: 0,
- queue: make(chan banktransfer.Transaction),
- retryQueue: make(chan banktransfer.Transaction),
- stop: make(chan struct{}),
- }
+ return &BankTransferService{keyGenerator: NewKeyGenerator()}
}
func (s *BankTransferService) TransferMoney(_ context.Context, transaction *banktransfer.Transaction) (*emptypb.Empty, error) {
@@ -41,76 +32,53 @@ func (s *BankTransferService) processTransaction(transaction *banktransfer.Trans
entry := log.WithField("transaction", transaction)
go func(transaction banktransfer.Transaction) {
entry.Info("Start processing transaction")
- transaction.Id = s.getUniqueId()
- time.Sleep(time.Duration(rand.Intn(9)+1) * time.Second)
- s.queue <- transaction
- entry.Info("Processing transaction finished")
+ transaction.Id = s.keyGenerator.getUniqueId()
+ if err := s.transactionWriter.Write(&transaction); err != nil {
+ entry.WithError(err).Error("Can't write transaction to transaction writer")
+ return
+ }
+ entry.Info("Transaction forwarded to transaction writer. Processing transaction finished")
}(*transaction)
}
-func (s *BankTransferService) getUniqueId() int32 {
- return atomic.AddInt32(&s.counter, 1)
-}
-
func (s *BankTransferService) ProcessTransactions(stream banktransfer.BankTransfer_ProcessTransactionsServer) error {
return func() error {
+ r := kafka.NewTransactionReader()
for {
- select {
- case <-stream.Context().Done():
- log.Info("Watching transactions cancelled from the client side")
- return nil
- case transaction := <-s.queue:
+ err := r.Read(stream.Context(), func(transaction *banktransfer.Transaction) error {
id := transaction.Id
- entry := log.WithField("transaction", transaction)
- entry.Info("Sending transaction")
- if err := stream.Send(&transaction); err != nil {
- s.requeueTransaction(&transaction)
- entry.WithError(err).Error("Error sending transaction")
- return err
+ if err := stream.Send(transaction); err != nil {
+ return fmt.Errorf("error sending transaction: %w", err)
}
- entry.Info("Transaction sent. Waiting for processing response")
response, err := stream.Recv()
if err != nil {
- s.requeueTransaction(&transaction)
- entry.WithError(err).Error("Error receiving processing response")
- return err
+ log.Errorf("error response transaction: %v", err)
}
if response.Id != id {
- // NOTE: this is just a guard and not happening as transaction is local per connection
- entry.Error("Received processing response of a different transaction")
- } else {
- entry.Info("Processing response received")
+ return errors.New("received processing response of a different transaction")
}
+ return nil
+ })
+ if err != nil {
+ log.Errorf("error reading transaction: %v", err)
+ break
}
}
+ if err := r.Close(); err != nil {
+ return err
+ }
+ return nil
}()
}
-func (s *BankTransferService) requeueTransaction(transaction *banktransfer.Transaction) {
- entry := log.WithField("transaction", transaction)
- go func(transaction banktransfer.Transaction) {
- entry.Infof("Requeuing transaction. Wait for %f seconds", retryTime.Seconds())
- time.Sleep(retryTime)
- s.retryQueue <- transaction
- entry.Info("Requeued transaction")
- }(*transaction)
-}
-
func (s *BankTransferService) Start() {
log.Info("Starting banktransfer service")
- go func() {
- for {
- select {
- case <-s.stop:
- break
- case transaction := <-s.retryQueue:
- s.queue <- transaction
- }
- }
- }()
+ s.transactionWriter = kafka.NewTransactionWriter()
+ log.Info("Successfully created transaction writer")
}
func (s *BankTransferService) Stop() {
log.Info("Stopping banktransfer service")
- close(s.stop)
+ s.transactionWriter.Close()
+ log.Info("Successfully closed connection to transaction writer")
}
diff --git a/src/banktransfer/service/keygenerator.go b/src/banktransfer/service/keygenerator.go
new file mode 100644
index 0000000000000000000000000000000000000000..fe400b215596d1fee38721a3cd2a2d426be9909f
--- /dev/null
+++ b/src/banktransfer/service/keygenerator.go
@@ -0,0 +1,26 @@
+package service
+
+import (
+ "fmt"
+ "os"
+ "sync/atomic"
+
+ log "github.com/sirupsen/logrus"
+)
+
+// KeyGenerator generates keys in the pattern {hostname}-{counter}
+type KeyGenerator struct {
+ counter int32
+ hostName string
+}
+
+func NewKeyGenerator() *KeyGenerator {
+ hostName, err := os.Hostname()
+ if err != nil {
+ log.Fatal(err)
+ }
+ return &KeyGenerator{counter: 0, hostName: hostName}
+}
+func (g *KeyGenerator) getUniqueId() string {
+ return fmt.Sprintf("%s-%d", g.hostName, atomic.AddInt32(&g.counter, 1))
+}
diff --git a/src/myaktion/docker-entrypoint.sh b/src/myaktion/docker-entrypoint.sh
index ab3b0d48e06de3d272f8e10365f2b125063ff2e0..29320a87f7d88262b90239f76d7fd169c46f0acd 100644
--- a/src/myaktion/docker-entrypoint.sh
+++ b/src/myaktion/docker-entrypoint.sh
@@ -4,12 +4,12 @@ set -e
# Wait for DB
if [ -n "$DB_CONNECT" ]; then
- /go/src/app/wait-for-it.sh "$DB_CONNECT" -t 20
+ /go/src/app/wait-for-it.sh "$DB_CONNECT" -t 120
fi
# Wait for banktransfer
if [ -n "$BANKTRANSFER_CONNECT" ]; then
- /go/src/app/wait-for-it.sh "$BANKTRANSFER_CONNECT" -t 20
+ /go/src/app/wait-for-it.sh "$BANKTRANSFER_CONNECT" -t 120
fi
# Run the main container command.