You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@dubbo.apache.org by GitBox <gi...@apache.org> on 2021/11/14 14:09:29 UTC
[GitHub] [dubbo-go-pixiu] tydhot opened a new pull request #300: [feature eventmesh]eventmesh
tydhot opened a new pull request #300:
URL: https://github.com/apache/dubbo-go-pixiu/pull/300
<!-- Thanks for sending a pull request!
-->
**What this PR does**:
**Which issue(s) this PR fixes**:
<!--
*Automatically closes linked issue when PR is merged.
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
_If PR is about `failing-tests or flakes`, please post the related issues/tests in a comment and do not use `Fixes`_*
-->
Fixes #
**Special notes for your reviewer**:
**Does this PR introduce a user-facing change?**:
<!--
If no, just write "NONE" in the release-note block below.
If yes, a release note is required:
Enter your extended release note in the block below. If the PR requires additional action from users switching to the new release, include the string "action required".
-->
```release-note
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go-pixiu] AlexStocks merged pull request #300: [feature eventmesh]eventmesh
Posted by GitBox <gi...@apache.org>.
AlexStocks merged pull request #300:
URL: https://github.com/apache/dubbo-go-pixiu/pull/300
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go-pixiu] codecov-commenter edited a comment on pull request #300: [feature eventmesh]eventmesh
Posted by GitBox <gi...@apache.org>.
codecov-commenter edited a comment on pull request #300:
URL: https://github.com/apache/dubbo-go-pixiu/pull/300#issuecomment-968298380
# [Codecov](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#300](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (89552f4) into [develop](https://codecov.io/gh/apache/dubbo-go-pixiu/commit/24d579c64a87a078b9374127092f3da2fc938a75?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (24d579c) will **not change** coverage.
> The diff coverage is `n/a`.
[](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## develop #300 +/- ##
========================================
Coverage 33.98% 33.98%
========================================
Files 61 61
Lines 3160 3160
========================================
Hits 1074 1074
Misses 1974 1974
Partials 112 112
```
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [24d579c...89552f4](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go-pixiu] tydhot commented on a change in pull request #300: [feature eventmesh]eventmesh
Posted by GitBox <gi...@apache.org>.
tydhot commented on a change in pull request #300:
URL: https://github.com/apache/dubbo-go-pixiu/pull/300#discussion_r754068323
##########
File path: pkg/client/mq/mq.go
##########
@@ -0,0 +1,156 @@
+/*
+ * 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 mq
+
+import (
+ "context"
+ "encoding/json"
+ "io/ioutil"
+ "strings"
+ "sync"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/client"
+ "github.com/apache/dubbo-go-pixiu/pkg/common/constant"
+ "github.com/apache/dubbo-go-pixiu/pkg/logger"
+)
+
+import (
+ perrors "github.com/pkg/errors"
+)
+
+var (
+ mqClient *Client
+ once sync.Once
+ consumerFacadeMap sync.Map
+)
+
+func NewSingletonMQClient(config Config) *Client {
+ if mqClient == nil {
+ once.Do(func() {
+ var err error
+ mqClient, err = NewMQClient(config)
+ if err != nil {
+ logger.Errorf("create mq client failed, %s", err.Error())
+ }
+ })
+ }
+ return mqClient
+}
+
+func NewMQClient(config Config) (*Client, error) {
+ var c *Client
+ ctx := context.Background()
+ switch config.MqType {
+ case constant.MQTypeKafka:
+ pf, err := NewKafkaProviderFacade(config.KafkaProducerConfig)
+ if err != nil {
+ return nil, err
+ }
+ c = &Client{
+ ctx: ctx,
+ producerFacade: pf,
+ kafkaConsumerConfig: config.KafkaConsumerConfig,
+ }
+ case constant.MQTypeRocketMQ:
+ return nil, perrors.New("rocketmq not support")
+ }
+
+ return c, nil
+}
+
+type Client struct {
+ ctx context.Context
+ producerFacade ProducerFacade
+ kafkaConsumerConfig KafkaConsumerConfig
+}
+
+func (c Client) Apply() error {
+ panic("implement me")
+}
+
+func (c Client) Close() error {
+ return nil
+}
+
+func (c Client) Call(req *client.Request) (res interface{}, err error) {
+ body, err := ioutil.ReadAll(req.IngressRequest.Body)
+ if err != nil {
+ return nil, err
+ }
+
+ paths := strings.Split(req.API.Path, "/")
+ if len(paths) < 3 {
+ return nil, perrors.New("failed to send message, broker or Topic not found")
+ }
+
+ switch MQActionStrToInt[paths[0]] {
+ case MQActionPublish:
+ var pReq MQProduceRequest
+ err = json.Unmarshal(body, &pReq)
+ if err != nil {
+ return nil, err
+ }
+ err = c.producerFacade.Send(pReq.Msg, WithTopic(pReq.Topic))
+ if err != nil {
+ return nil, err
+ }
+ case MQActionSubscribe:
+ var cReq MQSubscribeRequest
+ err = json.Unmarshal(body, &cReq)
+ if err != nil {
+ return nil, err
+ }
+ if _, ok := consumerFacadeMap.Load(cReq.ConsumerGroup); !ok {
Review comment:
这个场景我希望先简单化 订阅了之后解除订阅之前先不能改变订阅关系 后面在综合使用场景和各个mq之间的特性 再改变这里的逻辑 因为比如你先订阅A 再订阅B 那么这里最后的关系是AB都订阅呢 还是覆盖掉之前的A呢？还有 这里的topic关系修改对于各个mq的client api的改动可能都不尽的相同 如果把这些考虑上 整个业务逻辑的复杂度一下子就上升了 不如先从简
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go-pixiu] AlexStocks commented on a change in pull request #300: [feature eventmesh]eventmesh
Posted by GitBox <gi...@apache.org>.
AlexStocks commented on a change in pull request #300:
URL: https://github.com/apache/dubbo-go-pixiu/pull/300#discussion_r753845882
##########
File path: pkg/client/mq/config.go
##########
@@ -0,0 +1,62 @@
+/*
+ * 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 mq
+
+import (
+ "time"
+)
+
+type (
+ Config struct {
+ ClientId string `yaml:"client_id" json:"client_id"`
Review comment:
以后别再写 Id ，写作ID. ClientID
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go-pixiu] tydhot commented on a change in pull request #300: [feature eventmesh]eventmesh
Posted by GitBox <gi...@apache.org>.
tydhot commented on a change in pull request #300:
URL: https://github.com/apache/dubbo-go-pixiu/pull/300#discussion_r753804217
##########
File path: pkg/client/mq/kafka_facade.go
##########
@@ -0,0 +1,256 @@
+/*
+ * 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 mq
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "fmt"
+ "net/http"
+ "strconv"
+ "sync"
+ "time"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/logger"
+)
+
+import (
+ "github.com/Shopify/sarama"
+ perrors "github.com/pkg/errors"
+)
+
+type kafkaErrors struct {
+ count int
+ err string
+}
+
+func (ke kafkaErrors) Error() string {
+ return fmt.Sprintf("Failed to deliver %d messages due to %s", ke.count, ke.err)
+}
+
+func NewKafkaConsumerFacade(config KafkaConsumerConfig, consumerGroup string) (*KafkaConsumerFacade, error) {
+ c := sarama.NewConfig()
+ c.ClientID = config.ClientID
+ c.Metadata.Full = config.Metadata.Full
+ c.Metadata.Retry.Max = config.Metadata.Retry.Max
+ c.Metadata.Retry.Backoff = config.Metadata.Retry.Backoff
+ if config.ProtocolVersion != "" {
+ version, err := sarama.ParseKafkaVersion(config.ProtocolVersion)
+ if err != nil {
+ return nil, err
+ }
+ c.Version = version
+ }
+ client, err := sarama.NewConsumerGroup(config.Brokers, consumerGroup, c)
+ if err != nil {
+ return nil, err
+ }
+
+ return &KafkaConsumerFacade{consumerGroup: client, httpClient: &http.Client{Timeout: 5 * time.Second}, done: make(chan struct{})}, nil
+}
+
+type KafkaConsumerFacade struct {
+ consumerGroup sarama.ConsumerGroup
+ consumerManager map[string]func()
+ httpClient *http.Client
+ wg sync.WaitGroup
+ done chan struct{}
+}
+
+func (f *KafkaConsumerFacade) Subscribe(ctx context.Context, opts ...Option) error {
+ cOpt := DefaultOptions()
+ cOpt.ApplyOpts(opts...)
+ c, cancel := context.WithCancel(ctx)
+ key := GetConsumerManagerKey(cOpt.TopicList, cOpt.ConsumerGroup)
+ f.consumerManager[key] = cancel
+ f.wg.Add(2)
+ go f.consumeLoop(ctx, cOpt.TopicList, &consumerGroupHandler{cOpt.ConsumeUrl, f.httpClient})
+ go f.checkConsumerIsAlive(c, key, cOpt.CheckUrl)
+ return nil
+}
+
+func (f *KafkaConsumerFacade) consumeLoop(ctx context.Context, topics []string, handler sarama.ConsumerGroupHandler) {
+ for {
+ if _, ok := <-f.done; ok {
+ logger.Info("shutdown the consume loop")
+ break
+ }
+ if err := f.consumerGroup.Consume(ctx, topics, handler); err != nil {
+ logger.Warn("failed to consume the msg from kafka, %s", err.Error())
+ }
+ if ctx.Err() != nil {
+ // log consume stop
+ logger.Error("shutdown the consume loop due to %s", ctx.Err().Error())
+ break
+ }
+ }
+}
+
+type consumerGroupHandler struct {
+ consumerUrl string
+ httpClient *http.Client
+}
+
+func (c *consumerGroupHandler) Setup(session sarama.ConsumerGroupSession) error {
+ return nil
+}
+
+func (c *consumerGroupHandler) Cleanup(session sarama.ConsumerGroupSession) error {
+ return nil
+}
+
+func (c *consumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
+ for msg := range claim.Messages() {
+ session.MarkMessage(msg, "")
+ data, err := json.Marshal(MQMsgPush{Msg: []string{string(msg.Value)}})
+ if err != nil {
+ logger.Warn()
+ continue
+ }
+
+ req, err := http.NewRequest(http.MethodPost, c.consumerUrl, bytes.NewReader(data))
+ if err != nil {
+ logger.Warn()
+ continue
+ }
+ err = func() error {
+ resp, err := c.httpClient.Do(req)
+ if err != nil {
+ return err
+ }
+ defer resp.Body.Close()
+
+ if resp.StatusCode == http.StatusOK {
+ return nil
+ }
+ return perrors.New("failed send msg to consumer")
+ }()
+ if err != nil {
+ logger.Warn(err.Error())
+ }
+ }
+ return nil
+}
+
+// checkConsumerIsAlive make sure consumer is alive or would be removed from consumer list
+func (f *KafkaConsumerFacade) checkConsumerIsAlive(ctx context.Context, key string, checkUrl string) {
+ defer f.wg.Done()
+
+ ticker := time.NewTicker(15 * time.Second)
+ for {
+ select {
+ case <-f.done:
+ logger.Info()
+ case <-ticker.C:
+ lastCheck := 0
+ for i := 0; i < 5; i++ {
+ err := func() error {
+ req, err := http.NewRequest(http.MethodGet, checkUrl, bytes.NewReader([]byte{}))
+ if err != nil {
+ logger.Warn()
+ }
+
+ resp, err := f.httpClient.Do(req)
+ if err != nil {
+ logger.Warn()
+ }
+ defer resp.Body.Close()
+
+ lastCheck = resp.StatusCode
+ if resp.StatusCode != http.StatusOK {
+ return perrors.New("failed check consumer alive or not with status code " + strconv.Itoa(resp.StatusCode))
+ }
+
+ logger.Warn()
+ return nil
+ }()
+ if err != nil {
+ logger.Warn()
+ time.Sleep(10 * time.Millisecond)
+ } else {
+ break
+ }
+ }
+
+ if lastCheck != http.StatusOK {
+ f.consumerManager[key]()
+ delete(f.consumerManager, key)
+ }
+
+ }
+ }
+}
+
+func (f *KafkaConsumerFacade) UnSubscribe(opts ...Option) error {
+ return nil
+}
+
+func (f *KafkaConsumerFacade) Stop() {
+ close(f.done)
+ f.wg.Wait()
+}
+
+func NewKafkaProviderFacade(config KafkaProducerConfig) (*KafkaProducerFacade, error) {
+ c := sarama.NewConfig()
+ c.Producer.Return.Successes = true
+ c.Producer.Return.Errors = true
+ c.Producer.RequiredAcks = sarama.WaitForLocal
+ c.Metadata.Full = config.Metadata.Full
+ c.Metadata.Retry.Max = config.Metadata.Retry.Max
+ c.Metadata.Retry.Backoff = config.Metadata.Retry.Backoff
+ c.Producer.MaxMessageBytes = config.Producer.MaxMessageBytes
+ if config.ProtocolVersion != "" {
+ version, err := sarama.ParseKafkaVersion(config.ProtocolVersion)
+ if err != nil {
+ return nil, err
+ }
+ c.Version = version
+ }
+ producer, err := sarama.NewSyncProducer(config.Brokers, c)
+ if err != nil {
+ return nil, err
+ }
+ return &KafkaProducerFacade{producer: producer}, nil
+}
+
+type KafkaProducerFacade struct {
+ producer sarama.SyncProducer
+}
+
+func (k *KafkaProducerFacade) Send(msgs []string, opts ...Option) error {
+ pOpt := DefaultOptions()
+ pOpt.ApplyOpts(opts...)
+
+ pMsgs := make([]*sarama.ProducerMessage, 0)
Review comment:
一批消息在这里合并 发送到一个分区并不一定符合用户的预期 这里先分开 后面再提供参数来控制是否合并比较合适
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go-pixiu] ztelur commented on pull request #300: [feature eventmesh]eventmesh
Posted by GitBox <gi...@apache.org>.
ztelur commented on pull request #300:
URL: https://github.com/apache/dubbo-go-pixiu/pull/300#issuecomment-969914485
plz fix code issue in reviewdog comments
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go-pixiu] tydhot commented on a change in pull request #300: [feature eventmesh]eventmesh
Posted by GitBox <gi...@apache.org>.
tydhot commented on a change in pull request #300:
URL: https://github.com/apache/dubbo-go-pixiu/pull/300#discussion_r753896303
##########
File path: pkg/client/mq/config.go
##########
@@ -0,0 +1,62 @@
+/*
+ * 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 mq
+
+import (
+ "time"
+)
+
+type (
+ Config struct {
+ ClientId string `yaml:"client_id" json:"client_id"`
Review comment:
ok
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go-pixiu] tydhot commented on a change in pull request #300: [feature eventmesh]eventmesh
Posted by GitBox <gi...@apache.org>.
tydhot commented on a change in pull request #300:
URL: https://github.com/apache/dubbo-go-pixiu/pull/300#discussion_r753800307
##########
File path: pkg/client/mq/kafka_facade.go
##########
@@ -0,0 +1,256 @@
+/*
+ * 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 mq
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "fmt"
+ "net/http"
+ "strconv"
+ "sync"
+ "time"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/logger"
+)
+
+import (
+ "github.com/Shopify/sarama"
+ perrors "github.com/pkg/errors"
+)
+
+type kafkaErrors struct {
+ count int
+ err string
+}
+
+func (ke kafkaErrors) Error() string {
+ return fmt.Sprintf("Failed to deliver %d messages due to %s", ke.count, ke.err)
+}
+
+func NewKafkaConsumerFacade(config KafkaConsumerConfig, consumerGroup string) (*KafkaConsumerFacade, error) {
+ c := sarama.NewConfig()
+ c.ClientID = config.ClientID
+ c.Metadata.Full = config.Metadata.Full
+ c.Metadata.Retry.Max = config.Metadata.Retry.Max
+ c.Metadata.Retry.Backoff = config.Metadata.Retry.Backoff
+ if config.ProtocolVersion != "" {
+ version, err := sarama.ParseKafkaVersion(config.ProtocolVersion)
+ if err != nil {
+ return nil, err
+ }
+ c.Version = version
+ }
+ client, err := sarama.NewConsumerGroup(config.Brokers, consumerGroup, c)
+ if err != nil {
+ return nil, err
+ }
+
+ return &KafkaConsumerFacade{consumerGroup: client, httpClient: &http.Client{Timeout: 5 * time.Second}, done: make(chan struct{})}, nil
+}
+
+type KafkaConsumerFacade struct {
+ consumerGroup sarama.ConsumerGroup
+ consumerManager map[string]func()
+ httpClient *http.Client
+ wg sync.WaitGroup
+ done chan struct{}
+}
+
+func (f *KafkaConsumerFacade) Subscribe(ctx context.Context, opts ...Option) error {
+ cOpt := DefaultOptions()
+ cOpt.ApplyOpts(opts...)
+ c, cancel := context.WithCancel(ctx)
+ key := GetConsumerManagerKey(cOpt.TopicList, cOpt.ConsumerGroup)
+ f.consumerManager[key] = cancel
+ f.wg.Add(2)
+ go f.consumeLoop(ctx, cOpt.TopicList, &consumerGroupHandler{cOpt.ConsumeUrl, f.httpClient})
+ go f.checkConsumerIsAlive(c, key, cOpt.CheckUrl)
+ return nil
+}
+
+func (f *KafkaConsumerFacade) consumeLoop(ctx context.Context, topics []string, handler sarama.ConsumerGroupHandler) {
+ for {
+ if _, ok := <-f.done; ok {
Review comment:
之前是select， 是review dog建议改成这样
##########
File path: pkg/client/mq/mq.go
##########
@@ -0,0 +1,156 @@
+/*
+ * 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 mq
+
+import (
+ "context"
+ "encoding/json"
+ "io/ioutil"
+ "strings"
+ "sync"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/client"
+ "github.com/apache/dubbo-go-pixiu/pkg/common/constant"
+ "github.com/apache/dubbo-go-pixiu/pkg/logger"
+)
+
+import (
+ perrors "github.com/pkg/errors"
+)
+
+var (
+ mqClient *Client
+ once sync.Once
+ consumerFacadeMap sync.Map
+)
+
+func NewSingletonMQClient(config Config) *Client {
+ if mqClient == nil {
+ once.Do(func() {
+ var err error
+ mqClient, err = NewMQClient(config)
+ if err != nil {
+ logger.Errorf("create mq client failed, %s", err.Error())
+ }
+ })
+ }
+ return mqClient
+}
+
+func NewMQClient(config Config) (*Client, error) {
+ var c *Client
+ ctx := context.Background()
+ switch config.MqType {
+ case constant.MQTypeKafka:
+ pf, err := NewKafkaProviderFacade(config.KafkaProducerConfig)
+ if err != nil {
+ return nil, err
+ }
+ c = &Client{
+ ctx: ctx,
+ producerFacade: pf,
+ kafkaConsumerConfig: config.KafkaConsumerConfig,
+ }
+ case constant.MQTypeRocketMQ:
+ return nil, perrors.New("rocketmq not support")
+ }
+
+ return c, nil
+}
+
+type Client struct {
+ ctx context.Context
+ producerFacade ProducerFacade
+ kafkaConsumerConfig KafkaConsumerConfig
+}
+
+func (c Client) Apply() error {
+ panic("implement me")
+}
+
+func (c Client) Close() error {
+ return nil
+}
+
+func (c Client) Call(req *client.Request) (res interface{}, err error) {
+ body, err := ioutil.ReadAll(req.IngressRequest.Body)
+ if err != nil {
+ return nil, err
+ }
+
+ paths := strings.Split(req.API.Path, "/")
+ if len(paths) < 3 {
+ return nil, perrors.New("failed to send message, broker or Topic not found")
+ }
+
+ switch MQActionStrToInt[paths[0]] {
+ case MQActionPublish:
+ var pReq MQProduceRequest
+ err = json.Unmarshal(body, &pReq)
+ if err != nil {
+ return nil, err
+ }
+ err = c.producerFacade.Send(pReq.Msg, WithTopic(pReq.Topic))
+ if err != nil {
+ return nil, err
+ }
+ case MQActionSubscribe:
+ var cReq MQSubscribeRequest
+ err = json.Unmarshal(body, &cReq)
+ if err != nil {
+ return nil, err
+ }
+ if _, ok := consumerFacadeMap.Load(cReq.ConsumerGroup); !ok {
Review comment:
有数据说明已经订阅 考虑幂等性不要care了
##########
File path: pkg/client/mq/facade.go
##########
@@ -0,0 +1,92 @@
+/*
+ * 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 mq
+
+import (
+ "context"
+ "strings"
+)
+
+type ConsumerFacade interface {
+ // Subscribe message with specified broker and Topic, then handle msg with handler which send msg to real consumers
+ Subscribe(ctx context.Context, option ...Option) error
+ UnSubscribe(opts ...Option) error
+ Stop()
+}
+
+func GetConsumerManagerKey(topic []string, consumerGroup string) string {
+ return strings.Join((topic), ".") + "_" + consumerGroup
+}
+
+// MQOptions Consumer options
+// TODO: Add rocketmq params
+type MQOptions struct {
+ TopicList []string
+ ConsumeUrl string
+ CheckUrl string
+ ConsumerGroup string
+}
+
+func (o *MQOptions) ApplyOpts(opts ...Option) {
+ for _, opt := range opts {
+ opt(o)
+ }
+}
+
+func DefaultOptions() *MQOptions {
+ return &MQOptions{
+ TopicList: []string{"demo-topic"},
Review comment:
嗯 这个缺省项意义不大
##########
File path: pkg/client/mq/facade.go
##########
@@ -0,0 +1,92 @@
+/*
+ * 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 mq
+
+import (
+ "context"
+ "strings"
+)
+
+type ConsumerFacade interface {
+ // Subscribe message with specified broker and Topic, then handle msg with handler which send msg to real consumers
+ Subscribe(ctx context.Context, option ...Option) error
+ UnSubscribe(opts ...Option) error
+ Stop()
+}
+
+func GetConsumerManagerKey(topic []string, consumerGroup string) string {
+ return strings.Join((topic), ".") + "_" + consumerGroup
+}
+
+// MQOptions Consumer options
+// TODO: Add rocketmq params
+type MQOptions struct {
Review comment:
这个我觉得暂时问题不大 后面再简化也不迟
##########
File path: pkg/client/mq/mq.go
##########
@@ -0,0 +1,156 @@
+/*
+ * 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 mq
+
+import (
+ "context"
+ "encoding/json"
+ "io/ioutil"
+ "strings"
+ "sync"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/client"
+ "github.com/apache/dubbo-go-pixiu/pkg/common/constant"
+ "github.com/apache/dubbo-go-pixiu/pkg/logger"
+)
+
+import (
+ perrors "github.com/pkg/errors"
+)
+
+var (
+ mqClient *Client
+ once sync.Once
+ consumerFacadeMap sync.Map
+)
+
+func NewSingletonMQClient(config Config) *Client {
+ if mqClient == nil {
+ once.Do(func() {
+ var err error
+ mqClient, err = NewMQClient(config)
+ if err != nil {
+ logger.Errorf("create mq client failed, %s", err.Error())
+ }
+ })
+ }
+ return mqClient
+}
+
+func NewMQClient(config Config) (*Client, error) {
+ var c *Client
+ ctx := context.Background()
+ switch config.MqType {
+ case constant.MQTypeKafka:
+ pf, err := NewKafkaProviderFacade(config.KafkaProducerConfig)
+ if err != nil {
+ return nil, err
+ }
+ c = &Client{
+ ctx: ctx,
+ producerFacade: pf,
+ kafkaConsumerConfig: config.KafkaConsumerConfig,
+ }
+ case constant.MQTypeRocketMQ:
+ return nil, perrors.New("rocketmq not support")
+ }
+
+ return c, nil
+}
+
+type Client struct {
+ ctx context.Context
+ producerFacade ProducerFacade
+ kafkaConsumerConfig KafkaConsumerConfig
+}
+
+func (c Client) Apply() error {
+ panic("implement me")
+}
+
+func (c Client) Close() error {
+ return nil
+}
+
+func (c Client) Call(req *client.Request) (res interface{}, err error) {
+ body, err := ioutil.ReadAll(req.IngressRequest.Body)
+ if err != nil {
+ return nil, err
+ }
+
+ paths := strings.Split(req.API.Path, "/")
+ if len(paths) < 3 {
+ return nil, perrors.New("failed to send message, broker or Topic not found")
+ }
+
+ switch MQActionStrToInt[paths[0]] {
+ case MQActionPublish:
+ var pReq MQProduceRequest
+ err = json.Unmarshal(body, &pReq)
+ if err != nil {
+ return nil, err
+ }
+ err = c.producerFacade.Send(pReq.Msg, WithTopic(pReq.Topic))
+ if err != nil {
+ return nil, err
+ }
+ case MQActionSubscribe:
+ var cReq MQSubscribeRequest
+ err = json.Unmarshal(body, &cReq)
+ if err != nil {
+ return nil, err
+ }
+ if _, ok := consumerFacadeMap.Load(cReq.ConsumerGroup); !ok {
+ facade, err := NewKafkaConsumerFacade(c.kafkaConsumerConfig, cReq.ConsumerGroup)
+ if err != nil {
+ return nil, err
+ }
+ consumerFacadeMap.Store(cReq.ConsumerGroup, facade)
+ if f, ok := consumerFacadeMap.Load(cReq.ConsumerGroup); ok {
Review comment:
consumerFacadeMap不是局部变量 我觉得这样比较好
##########
File path: pkg/client/mq/kafka_facade.go
##########
@@ -0,0 +1,256 @@
+/*
+ * 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 mq
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "fmt"
+ "net/http"
+ "strconv"
+ "sync"
+ "time"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/logger"
+)
+
+import (
+ "github.com/Shopify/sarama"
+ perrors "github.com/pkg/errors"
+)
+
+type kafkaErrors struct {
+ count int
+ err string
+}
+
+func (ke kafkaErrors) Error() string {
+ return fmt.Sprintf("Failed to deliver %d messages due to %s", ke.count, ke.err)
+}
+
+func NewKafkaConsumerFacade(config KafkaConsumerConfig, consumerGroup string) (*KafkaConsumerFacade, error) {
+ c := sarama.NewConfig()
+ c.ClientID = config.ClientID
+ c.Metadata.Full = config.Metadata.Full
+ c.Metadata.Retry.Max = config.Metadata.Retry.Max
+ c.Metadata.Retry.Backoff = config.Metadata.Retry.Backoff
+ if config.ProtocolVersion != "" {
+ version, err := sarama.ParseKafkaVersion(config.ProtocolVersion)
+ if err != nil {
+ return nil, err
+ }
+ c.Version = version
+ }
+ client, err := sarama.NewConsumerGroup(config.Brokers, consumerGroup, c)
+ if err != nil {
+ return nil, err
+ }
+
+ return &KafkaConsumerFacade{consumerGroup: client, httpClient: &http.Client{Timeout: 5 * time.Second}, done: make(chan struct{})}, nil
+}
+
+type KafkaConsumerFacade struct {
+ consumerGroup sarama.ConsumerGroup
+ consumerManager map[string]func()
+ httpClient *http.Client
+ wg sync.WaitGroup
+ done chan struct{}
+}
+
+func (f *KafkaConsumerFacade) Subscribe(ctx context.Context, opts ...Option) error {
+ cOpt := DefaultOptions()
+ cOpt.ApplyOpts(opts...)
+ c, cancel := context.WithCancel(ctx)
+ key := GetConsumerManagerKey(cOpt.TopicList, cOpt.ConsumerGroup)
+ f.consumerManager[key] = cancel
+ f.wg.Add(2)
+ go f.consumeLoop(ctx, cOpt.TopicList, &consumerGroupHandler{cOpt.ConsumeUrl, f.httpClient})
+ go f.checkConsumerIsAlive(c, key, cOpt.CheckUrl)
+ return nil
+}
+
+func (f *KafkaConsumerFacade) consumeLoop(ctx context.Context, topics []string, handler sarama.ConsumerGroupHandler) {
+ for {
+ if _, ok := <-f.done; ok {
+ logger.Info("shutdown the consume loop")
+ break
+ }
+ if err := f.consumerGroup.Consume(ctx, topics, handler); err != nil {
+ logger.Warn("failed to consume the msg from kafka, %s", err.Error())
+ }
+ if ctx.Err() != nil {
+ // log consume stop
+ logger.Error("shutdown the consume loop due to %s", ctx.Err().Error())
+ break
+ }
+ }
+}
+
+type consumerGroupHandler struct {
+ consumerUrl string
+ httpClient *http.Client
+}
+
+func (c *consumerGroupHandler) Setup(session sarama.ConsumerGroupSession) error {
+ return nil
+}
+
+func (c *consumerGroupHandler) Cleanup(session sarama.ConsumerGroupSession) error {
+ return nil
+}
+
+func (c *consumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
+ for msg := range claim.Messages() {
+ session.MarkMessage(msg, "")
+ data, err := json.Marshal(MQMsgPush{Msg: []string{string(msg.Value)}})
+ if err != nil {
+ logger.Warn()
+ continue
+ }
+
+ req, err := http.NewRequest(http.MethodPost, c.consumerUrl, bytes.NewReader(data))
+ if err != nil {
+ logger.Warn()
+ continue
+ }
+ err = func() error {
+ resp, err := c.httpClient.Do(req)
+ if err != nil {
+ return err
+ }
+ defer resp.Body.Close()
+
+ if resp.StatusCode == http.StatusOK {
+ return nil
+ }
+ return perrors.New("failed send msg to consumer")
+ }()
+ if err != nil {
+ logger.Warn(err.Error())
+ }
+ }
+ return nil
+}
+
+// checkConsumerIsAlive make sure consumer is alive or would be removed from consumer list
+func (f *KafkaConsumerFacade) checkConsumerIsAlive(ctx context.Context, key string, checkUrl string) {
+ defer f.wg.Done()
+
+ ticker := time.NewTicker(15 * time.Second)
+ for {
+ select {
+ case <-f.done:
+ logger.Info()
Review comment:
ok
##########
File path: pkg/client/mq/kafka_facade.go
##########
@@ -0,0 +1,256 @@
+/*
+ * 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 mq
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "fmt"
+ "net/http"
+ "strconv"
+ "sync"
+ "time"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/logger"
+)
+
+import (
+ "github.com/Shopify/sarama"
+ perrors "github.com/pkg/errors"
+)
+
+type kafkaErrors struct {
+ count int
+ err string
+}
+
+func (ke kafkaErrors) Error() string {
+ return fmt.Sprintf("Failed to deliver %d messages due to %s", ke.count, ke.err)
+}
+
+func NewKafkaConsumerFacade(config KafkaConsumerConfig, consumerGroup string) (*KafkaConsumerFacade, error) {
+ c := sarama.NewConfig()
+ c.ClientID = config.ClientID
+ c.Metadata.Full = config.Metadata.Full
+ c.Metadata.Retry.Max = config.Metadata.Retry.Max
+ c.Metadata.Retry.Backoff = config.Metadata.Retry.Backoff
+ if config.ProtocolVersion != "" {
+ version, err := sarama.ParseKafkaVersion(config.ProtocolVersion)
+ if err != nil {
+ return nil, err
+ }
+ c.Version = version
+ }
+ client, err := sarama.NewConsumerGroup(config.Brokers, consumerGroup, c)
+ if err != nil {
+ return nil, err
+ }
+
+ return &KafkaConsumerFacade{consumerGroup: client, httpClient: &http.Client{Timeout: 5 * time.Second}, done: make(chan struct{})}, nil
+}
+
+type KafkaConsumerFacade struct {
+ consumerGroup sarama.ConsumerGroup
+ consumerManager map[string]func()
+ httpClient *http.Client
+ wg sync.WaitGroup
+ done chan struct{}
+}
+
+func (f *KafkaConsumerFacade) Subscribe(ctx context.Context, opts ...Option) error {
+ cOpt := DefaultOptions()
+ cOpt.ApplyOpts(opts...)
+ c, cancel := context.WithCancel(ctx)
+ key := GetConsumerManagerKey(cOpt.TopicList, cOpt.ConsumerGroup)
+ f.consumerManager[key] = cancel
+ f.wg.Add(2)
+ go f.consumeLoop(ctx, cOpt.TopicList, &consumerGroupHandler{cOpt.ConsumeUrl, f.httpClient})
+ go f.checkConsumerIsAlive(c, key, cOpt.CheckUrl)
+ return nil
+}
+
+func (f *KafkaConsumerFacade) consumeLoop(ctx context.Context, topics []string, handler sarama.ConsumerGroupHandler) {
+ for {
+ if _, ok := <-f.done; ok {
+ logger.Info("shutdown the consume loop")
+ break
+ }
+ if err := f.consumerGroup.Consume(ctx, topics, handler); err != nil {
+ logger.Warn("failed to consume the msg from kafka, %s", err.Error())
+ }
+ if ctx.Err() != nil {
+ // log consume stop
+ logger.Error("shutdown the consume loop due to %s", ctx.Err().Error())
+ break
+ }
+ }
+}
+
+type consumerGroupHandler struct {
+ consumerUrl string
+ httpClient *http.Client
+}
+
+func (c *consumerGroupHandler) Setup(session sarama.ConsumerGroupSession) error {
+ return nil
+}
+
+func (c *consumerGroupHandler) Cleanup(session sarama.ConsumerGroupSession) error {
+ return nil
+}
+
+func (c *consumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
+ for msg := range claim.Messages() {
+ session.MarkMessage(msg, "")
+ data, err := json.Marshal(MQMsgPush{Msg: []string{string(msg.Value)}})
+ if err != nil {
+ logger.Warn()
+ continue
+ }
+
+ req, err := http.NewRequest(http.MethodPost, c.consumerUrl, bytes.NewReader(data))
+ if err != nil {
+ logger.Warn()
+ continue
+ }
+ err = func() error {
+ resp, err := c.httpClient.Do(req)
+ if err != nil {
+ return err
+ }
+ defer resp.Body.Close()
+
+ if resp.StatusCode == http.StatusOK {
+ return nil
+ }
+ return perrors.New("failed send msg to consumer")
+ }()
+ if err != nil {
+ logger.Warn(err.Error())
+ }
+ }
+ return nil
+}
+
+// checkConsumerIsAlive make sure consumer is alive or would be removed from consumer list
+func (f *KafkaConsumerFacade) checkConsumerIsAlive(ctx context.Context, key string, checkUrl string) {
+ defer f.wg.Done()
+
+ ticker := time.NewTicker(15 * time.Second)
+ for {
+ select {
+ case <-f.done:
+ logger.Info()
+ case <-ticker.C:
+ lastCheck := 0
+ for i := 0; i < 5; i++ {
+ err := func() error {
+ req, err := http.NewRequest(http.MethodGet, checkUrl, bytes.NewReader([]byte{}))
+ if err != nil {
+ logger.Warn()
+ }
+
+ resp, err := f.httpClient.Do(req)
+ if err != nil {
+ logger.Warn()
+ }
+ defer resp.Body.Close()
+
+ lastCheck = resp.StatusCode
+ if resp.StatusCode != http.StatusOK {
+ return perrors.New("failed check consumer alive or not with status code " + strconv.Itoa(resp.StatusCode))
+ }
+
+ logger.Warn()
+ return nil
+ }()
+ if err != nil {
+ logger.Warn()
+ time.Sleep(10 * time.Millisecond)
+ } else {
+ break
+ }
+ }
+
+ if lastCheck != http.StatusOK {
+ f.consumerManager[key]()
+ delete(f.consumerManager, key)
+ }
+
+ }
+ }
+}
+
+func (f *KafkaConsumerFacade) UnSubscribe(opts ...Option) error {
+ return nil
+}
+
+func (f *KafkaConsumerFacade) Stop() {
+ close(f.done)
+ f.wg.Wait()
+}
+
+func NewKafkaProviderFacade(config KafkaProducerConfig) (*KafkaProducerFacade, error) {
+ c := sarama.NewConfig()
+ c.Producer.Return.Successes = true
+ c.Producer.Return.Errors = true
+ c.Producer.RequiredAcks = sarama.WaitForLocal
+ c.Metadata.Full = config.Metadata.Full
+ c.Metadata.Retry.Max = config.Metadata.Retry.Max
+ c.Metadata.Retry.Backoff = config.Metadata.Retry.Backoff
+ c.Producer.MaxMessageBytes = config.Producer.MaxMessageBytes
+ if config.ProtocolVersion != "" {
+ version, err := sarama.ParseKafkaVersion(config.ProtocolVersion)
+ if err != nil {
+ return nil, err
+ }
+ c.Version = version
+ }
+ producer, err := sarama.NewSyncProducer(config.Brokers, c)
+ if err != nil {
+ return nil, err
+ }
+ return &KafkaProducerFacade{producer: producer}, nil
+}
+
+type KafkaProducerFacade struct {
+ producer sarama.SyncProducer
+}
+
+func (k *KafkaProducerFacade) Send(msgs []string, opts ...Option) error {
+ pOpt := DefaultOptions()
+ pOpt.ApplyOpts(opts...)
+
+ pMsgs := make([]*sarama.ProducerMessage, 0)
Review comment:
一批消息在这里合并 发送到一个分区并不一定符合用户的预期 这里先分开 后面再提供参数来控制是否合并比较合适
##########
File path: pkg/client/mq/kafka_facade.go
##########
@@ -0,0 +1,256 @@
+/*
+ * 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 mq
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "fmt"
+ "net/http"
+ "strconv"
+ "sync"
+ "time"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/logger"
+)
+
+import (
+ "github.com/Shopify/sarama"
+ perrors "github.com/pkg/errors"
+)
+
+type kafkaErrors struct {
+ count int
+ err string
+}
+
+func (ke kafkaErrors) Error() string {
+ return fmt.Sprintf("Failed to deliver %d messages due to %s", ke.count, ke.err)
+}
+
+func NewKafkaConsumerFacade(config KafkaConsumerConfig, consumerGroup string) (*KafkaConsumerFacade, error) {
+ c := sarama.NewConfig()
+ c.ClientID = config.ClientID
+ c.Metadata.Full = config.Metadata.Full
+ c.Metadata.Retry.Max = config.Metadata.Retry.Max
+ c.Metadata.Retry.Backoff = config.Metadata.Retry.Backoff
+ if config.ProtocolVersion != "" {
+ version, err := sarama.ParseKafkaVersion(config.ProtocolVersion)
+ if err != nil {
+ return nil, err
+ }
+ c.Version = version
+ }
+ client, err := sarama.NewConsumerGroup(config.Brokers, consumerGroup, c)
+ if err != nil {
+ return nil, err
+ }
+
+ return &KafkaConsumerFacade{consumerGroup: client, httpClient: &http.Client{Timeout: 5 * time.Second}, done: make(chan struct{})}, nil
+}
+
+type KafkaConsumerFacade struct {
+ consumerGroup sarama.ConsumerGroup
+ consumerManager map[string]func()
+ httpClient *http.Client
+ wg sync.WaitGroup
+ done chan struct{}
+}
+
+func (f *KafkaConsumerFacade) Subscribe(ctx context.Context, opts ...Option) error {
+ cOpt := DefaultOptions()
+ cOpt.ApplyOpts(opts...)
+ c, cancel := context.WithCancel(ctx)
+ key := GetConsumerManagerKey(cOpt.TopicList, cOpt.ConsumerGroup)
+ f.consumerManager[key] = cancel
+ f.wg.Add(2)
+ go f.consumeLoop(ctx, cOpt.TopicList, &consumerGroupHandler{cOpt.ConsumeUrl, f.httpClient})
+ go f.checkConsumerIsAlive(c, key, cOpt.CheckUrl)
+ return nil
+}
+
+func (f *KafkaConsumerFacade) consumeLoop(ctx context.Context, topics []string, handler sarama.ConsumerGroupHandler) {
+ for {
+ if _, ok := <-f.done; ok {
+ logger.Info("shutdown the consume loop")
+ break
+ }
+ if err := f.consumerGroup.Consume(ctx, topics, handler); err != nil {
+ logger.Warn("failed to consume the msg from kafka, %s", err.Error())
+ }
+ if ctx.Err() != nil {
+ // log consume stop
+ logger.Error("shutdown the consume loop due to %s", ctx.Err().Error())
+ break
+ }
+ }
+}
+
+type consumerGroupHandler struct {
+ consumerUrl string
+ httpClient *http.Client
+}
+
+func (c *consumerGroupHandler) Setup(session sarama.ConsumerGroupSession) error {
+ return nil
+}
+
+func (c *consumerGroupHandler) Cleanup(session sarama.ConsumerGroupSession) error {
+ return nil
+}
+
+func (c *consumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
+ for msg := range claim.Messages() {
+ session.MarkMessage(msg, "")
+ data, err := json.Marshal(MQMsgPush{Msg: []string{string(msg.Value)}})
+ if err != nil {
+ logger.Warn()
+ continue
+ }
+
+ req, err := http.NewRequest(http.MethodPost, c.consumerUrl, bytes.NewReader(data))
+ if err != nil {
+ logger.Warn()
+ continue
+ }
+ err = func() error {
+ resp, err := c.httpClient.Do(req)
+ if err != nil {
+ return err
+ }
+ defer resp.Body.Close()
+
+ if resp.StatusCode == http.StatusOK {
+ return nil
+ }
+ return perrors.New("failed send msg to consumer")
+ }()
+ if err != nil {
+ logger.Warn(err.Error())
+ }
+ }
+ return nil
+}
+
+// checkConsumerIsAlive make sure consumer is alive or would be removed from consumer list
+func (f *KafkaConsumerFacade) checkConsumerIsAlive(ctx context.Context, key string, checkUrl string) {
+ defer f.wg.Done()
+
+ ticker := time.NewTicker(15 * time.Second)
Review comment:
ok
##########
File path: pkg/client/mq/kafka_facade.go
##########
@@ -0,0 +1,256 @@
+/*
+ * 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 mq
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "fmt"
+ "net/http"
+ "strconv"
+ "sync"
+ "time"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/logger"
+)
+
+import (
+ "github.com/Shopify/sarama"
+ perrors "github.com/pkg/errors"
+)
+
+type kafkaErrors struct {
+ count int
+ err string
+}
+
+func (ke kafkaErrors) Error() string {
+ return fmt.Sprintf("Failed to deliver %d messages due to %s", ke.count, ke.err)
+}
+
+func NewKafkaConsumerFacade(config KafkaConsumerConfig, consumerGroup string) (*KafkaConsumerFacade, error) {
+ c := sarama.NewConfig()
+ c.ClientID = config.ClientID
+ c.Metadata.Full = config.Metadata.Full
+ c.Metadata.Retry.Max = config.Metadata.Retry.Max
+ c.Metadata.Retry.Backoff = config.Metadata.Retry.Backoff
+ if config.ProtocolVersion != "" {
+ version, err := sarama.ParseKafkaVersion(config.ProtocolVersion)
+ if err != nil {
+ return nil, err
+ }
+ c.Version = version
+ }
+ client, err := sarama.NewConsumerGroup(config.Brokers, consumerGroup, c)
+ if err != nil {
+ return nil, err
+ }
+
+ return &KafkaConsumerFacade{consumerGroup: client, httpClient: &http.Client{Timeout: 5 * time.Second}, done: make(chan struct{})}, nil
+}
+
+type KafkaConsumerFacade struct {
+ consumerGroup sarama.ConsumerGroup
+ consumerManager map[string]func()
+ httpClient *http.Client
+ wg sync.WaitGroup
+ done chan struct{}
+}
+
+func (f *KafkaConsumerFacade) Subscribe(ctx context.Context, opts ...Option) error {
+ cOpt := DefaultOptions()
+ cOpt.ApplyOpts(opts...)
+ c, cancel := context.WithCancel(ctx)
+ key := GetConsumerManagerKey(cOpt.TopicList, cOpt.ConsumerGroup)
Review comment:
这里我和俊铭的考虑是 这一块其实是在kafka之上的一个通用的facade抽象。目前因为只有一个kafka实现可能比较冗余，但是在多种实现下，这里可能需要把一些通用参数转化成每种mq能够进行必要性判断和具体使用的格式，这里主要是出于这样 的考虑不过MQOptions也应该移动到kafka_facade内
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go-pixiu] codecov-commenter edited a comment on pull request #300: [feature eventmesh]eventmesh
Posted by GitBox <gi...@apache.org>.
codecov-commenter edited a comment on pull request #300:
URL: https://github.com/apache/dubbo-go-pixiu/pull/300#issuecomment-968298380
# [Codecov](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#300](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (aa2ada5) into [develop](https://codecov.io/gh/apache/dubbo-go-pixiu/commit/bf6dde05c14d394a5f6bda919fc6473107ae8d25?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (bf6dde0) will **not change** coverage.
> The diff coverage is `n/a`.
[](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## develop #300 +/- ##
========================================
Coverage 36.04% 36.04%
========================================
Files 55 55
Lines 3299 3299
========================================
Hits 1189 1189
Misses 1991 1991
Partials 119 119
```
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [bf6dde0...aa2ada5](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go-pixiu] codecov-commenter edited a comment on pull request #300: [feature eventmesh]eventmesh
Posted by GitBox <gi...@apache.org>.
codecov-commenter edited a comment on pull request #300:
URL: https://github.com/apache/dubbo-go-pixiu/pull/300#issuecomment-968298380
# [Codecov](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#300](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (3c13d5b) into [develop](https://codecov.io/gh/apache/dubbo-go-pixiu/commit/5bef4853016d8bfb0da4db715ccc5480cbcde7a6?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (5bef485) will **not change** coverage.
> The diff coverage is `n/a`.
[](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## develop #300 +/- ##
========================================
Coverage 33.93% 33.93%
========================================
Files 61 61
Lines 3165 3165
========================================
Hits 1074 1074
Misses 1979 1979
Partials 112 112
```
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [5bef485...3c13d5b](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go-pixiu] codecov-commenter edited a comment on pull request #300: [feature eventmesh]eventmesh
Posted by GitBox <gi...@apache.org>.
codecov-commenter edited a comment on pull request #300:
URL: https://github.com/apache/dubbo-go-pixiu/pull/300#issuecomment-968298380
# [Codecov](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#300](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (f06e488) into [develop](https://codecov.io/gh/apache/dubbo-go-pixiu/commit/24d579c64a87a078b9374127092f3da2fc938a75?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (24d579c) will **not change** coverage.
> The diff coverage is `n/a`.
[](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## develop #300 +/- ##
========================================
Coverage 33.98% 33.98%
========================================
Files 61 61
Lines 3160 3160
========================================
Hits 1074 1074
Misses 1974 1974
Partials 112 112
```
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [24d579c...f06e488](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go-pixiu] codecov-commenter commented on pull request #300: [feature eventmesh]eventmesh
Posted by GitBox <gi...@apache.org>.
codecov-commenter commented on pull request #300:
URL: https://github.com/apache/dubbo-go-pixiu/pull/300#issuecomment-968298380
# [Codecov](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#300](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (89552f4) into [develop](https://codecov.io/gh/apache/dubbo-go-pixiu/commit/24d579c64a87a078b9374127092f3da2fc938a75?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (24d579c) will **not change** coverage.
> The diff coverage is `n/a`.
[](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## develop #300 +/- ##
========================================
Coverage 33.98% 33.98%
========================================
Files 61 61
Lines 3160 3160
========================================
Hits 1074 1074
Misses 1974 1974
Partials 112 112
```
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [24d579c...89552f4](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go-pixiu] tydhot commented on a change in pull request #300: [feature eventmesh]eventmesh
Posted by GitBox <gi...@apache.org>.
tydhot commented on a change in pull request #300:
URL: https://github.com/apache/dubbo-go-pixiu/pull/300#discussion_r753803615
##########
File path: pkg/client/mq/kafka_facade.go
##########
@@ -0,0 +1,256 @@
+/*
+ * 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 mq
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "fmt"
+ "net/http"
+ "strconv"
+ "sync"
+ "time"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/logger"
+)
+
+import (
+ "github.com/Shopify/sarama"
+ perrors "github.com/pkg/errors"
+)
+
+type kafkaErrors struct {
+ count int
+ err string
+}
+
+func (ke kafkaErrors) Error() string {
+ return fmt.Sprintf("Failed to deliver %d messages due to %s", ke.count, ke.err)
+}
+
+func NewKafkaConsumerFacade(config KafkaConsumerConfig, consumerGroup string) (*KafkaConsumerFacade, error) {
+ c := sarama.NewConfig()
+ c.ClientID = config.ClientID
+ c.Metadata.Full = config.Metadata.Full
+ c.Metadata.Retry.Max = config.Metadata.Retry.Max
+ c.Metadata.Retry.Backoff = config.Metadata.Retry.Backoff
+ if config.ProtocolVersion != "" {
+ version, err := sarama.ParseKafkaVersion(config.ProtocolVersion)
+ if err != nil {
+ return nil, err
+ }
+ c.Version = version
+ }
+ client, err := sarama.NewConsumerGroup(config.Brokers, consumerGroup, c)
+ if err != nil {
+ return nil, err
+ }
+
+ return &KafkaConsumerFacade{consumerGroup: client, httpClient: &http.Client{Timeout: 5 * time.Second}, done: make(chan struct{})}, nil
+}
+
+type KafkaConsumerFacade struct {
+ consumerGroup sarama.ConsumerGroup
+ consumerManager map[string]func()
+ httpClient *http.Client
+ wg sync.WaitGroup
+ done chan struct{}
+}
+
+func (f *KafkaConsumerFacade) Subscribe(ctx context.Context, opts ...Option) error {
+ cOpt := DefaultOptions()
+ cOpt.ApplyOpts(opts...)
+ c, cancel := context.WithCancel(ctx)
+ key := GetConsumerManagerKey(cOpt.TopicList, cOpt.ConsumerGroup)
+ f.consumerManager[key] = cancel
+ f.wg.Add(2)
+ go f.consumeLoop(ctx, cOpt.TopicList, &consumerGroupHandler{cOpt.ConsumeUrl, f.httpClient})
+ go f.checkConsumerIsAlive(c, key, cOpt.CheckUrl)
+ return nil
+}
+
+func (f *KafkaConsumerFacade) consumeLoop(ctx context.Context, topics []string, handler sarama.ConsumerGroupHandler) {
+ for {
+ if _, ok := <-f.done; ok {
+ logger.Info("shutdown the consume loop")
+ break
+ }
+ if err := f.consumerGroup.Consume(ctx, topics, handler); err != nil {
+ logger.Warn("failed to consume the msg from kafka, %s", err.Error())
+ }
+ if ctx.Err() != nil {
+ // log consume stop
+ logger.Error("shutdown the consume loop due to %s", ctx.Err().Error())
+ break
+ }
+ }
+}
+
+type consumerGroupHandler struct {
+ consumerUrl string
+ httpClient *http.Client
+}
+
+func (c *consumerGroupHandler) Setup(session sarama.ConsumerGroupSession) error {
+ return nil
+}
+
+func (c *consumerGroupHandler) Cleanup(session sarama.ConsumerGroupSession) error {
+ return nil
+}
+
+func (c *consumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
+ for msg := range claim.Messages() {
+ session.MarkMessage(msg, "")
+ data, err := json.Marshal(MQMsgPush{Msg: []string{string(msg.Value)}})
+ if err != nil {
+ logger.Warn()
+ continue
+ }
+
+ req, err := http.NewRequest(http.MethodPost, c.consumerUrl, bytes.NewReader(data))
+ if err != nil {
+ logger.Warn()
+ continue
+ }
+ err = func() error {
+ resp, err := c.httpClient.Do(req)
+ if err != nil {
+ return err
+ }
+ defer resp.Body.Close()
+
+ if resp.StatusCode == http.StatusOK {
+ return nil
+ }
+ return perrors.New("failed send msg to consumer")
+ }()
+ if err != nil {
+ logger.Warn(err.Error())
+ }
+ }
+ return nil
+}
+
+// checkConsumerIsAlive make sure consumer is alive or would be removed from consumer list
+func (f *KafkaConsumerFacade) checkConsumerIsAlive(ctx context.Context, key string, checkUrl string) {
+ defer f.wg.Done()
+
+ ticker := time.NewTicker(15 * time.Second)
+ for {
+ select {
+ case <-f.done:
+ logger.Info()
Review comment:
ok
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go-pixiu] tydhot commented on a change in pull request #300: [feature eventmesh]eventmesh
Posted by GitBox <gi...@apache.org>.
tydhot commented on a change in pull request #300:
URL: https://github.com/apache/dubbo-go-pixiu/pull/300#discussion_r753803359
##########
File path: pkg/client/mq/mq.go
##########
@@ -0,0 +1,156 @@
+/*
+ * 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 mq
+
+import (
+ "context"
+ "encoding/json"
+ "io/ioutil"
+ "strings"
+ "sync"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/client"
+ "github.com/apache/dubbo-go-pixiu/pkg/common/constant"
+ "github.com/apache/dubbo-go-pixiu/pkg/logger"
+)
+
+import (
+ perrors "github.com/pkg/errors"
+)
+
+var (
+ mqClient *Client
+ once sync.Once
+ consumerFacadeMap sync.Map
+)
+
+func NewSingletonMQClient(config Config) *Client {
+ if mqClient == nil {
+ once.Do(func() {
+ var err error
+ mqClient, err = NewMQClient(config)
+ if err != nil {
+ logger.Errorf("create mq client failed, %s", err.Error())
+ }
+ })
+ }
+ return mqClient
+}
+
+func NewMQClient(config Config) (*Client, error) {
+ var c *Client
+ ctx := context.Background()
+ switch config.MqType {
+ case constant.MQTypeKafka:
+ pf, err := NewKafkaProviderFacade(config.KafkaProducerConfig)
+ if err != nil {
+ return nil, err
+ }
+ c = &Client{
+ ctx: ctx,
+ producerFacade: pf,
+ kafkaConsumerConfig: config.KafkaConsumerConfig,
+ }
+ case constant.MQTypeRocketMQ:
+ return nil, perrors.New("rocketmq not support")
+ }
+
+ return c, nil
+}
+
+type Client struct {
+ ctx context.Context
+ producerFacade ProducerFacade
+ kafkaConsumerConfig KafkaConsumerConfig
+}
+
+func (c Client) Apply() error {
+ panic("implement me")
+}
+
+func (c Client) Close() error {
+ return nil
+}
+
+func (c Client) Call(req *client.Request) (res interface{}, err error) {
+ body, err := ioutil.ReadAll(req.IngressRequest.Body)
+ if err != nil {
+ return nil, err
+ }
+
+ paths := strings.Split(req.API.Path, "/")
+ if len(paths) < 3 {
+ return nil, perrors.New("failed to send message, broker or Topic not found")
+ }
+
+ switch MQActionStrToInt[paths[0]] {
+ case MQActionPublish:
+ var pReq MQProduceRequest
+ err = json.Unmarshal(body, &pReq)
+ if err != nil {
+ return nil, err
+ }
+ err = c.producerFacade.Send(pReq.Msg, WithTopic(pReq.Topic))
+ if err != nil {
+ return nil, err
+ }
+ case MQActionSubscribe:
+ var cReq MQSubscribeRequest
+ err = json.Unmarshal(body, &cReq)
+ if err != nil {
+ return nil, err
+ }
+ if _, ok := consumerFacadeMap.Load(cReq.ConsumerGroup); !ok {
+ facade, err := NewKafkaConsumerFacade(c.kafkaConsumerConfig, cReq.ConsumerGroup)
+ if err != nil {
+ return nil, err
+ }
+ consumerFacadeMap.Store(cReq.ConsumerGroup, facade)
+ if f, ok := consumerFacadeMap.Load(cReq.ConsumerGroup); ok {
Review comment:
consumerFacadeMap不是局部变量 我觉得这样比较好
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go-pixiu] codecov-commenter edited a comment on pull request #300: [feature eventmesh]eventmesh
Posted by GitBox <gi...@apache.org>.
codecov-commenter edited a comment on pull request #300:
URL: https://github.com/apache/dubbo-go-pixiu/pull/300#issuecomment-968298380
# [Codecov](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#300](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (84c4282) into [develop](https://codecov.io/gh/apache/dubbo-go-pixiu/commit/ec68432b71b733a1d87b7a182ded2004fed073d9?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (ec68432) will **decrease** coverage by `0.05%`.
> The diff coverage is `n/a`.
[](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## develop #300 +/- ##
===========================================
- Coverage 33.98% 33.93% -0.06%
===========================================
Files 61 61
Lines 3160 3165 +5
===========================================
Hits 1074 1074
- Misses 1974 1979 +5
Partials 112 112
```
| [Impacted Files](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [pkg/model/base.go](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGtnL21vZGVsL2Jhc2UuZ28=) | `0.00% <0.00%> (ø)` | |
| [pkg/model/http.go](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGtnL21vZGVsL2h0dHAuZ28=) | `0.00% <0.00%> (ø)` | |
| [pkg/server/listener.go](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGtnL3NlcnZlci9saXN0ZW5lci5nbw==) | `0.00% <0.00%> (ø)` | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [ec68432...84c4282](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go-pixiu] ztelur commented on a change in pull request #300: [feature eventmesh]eventmesh
Posted by GitBox <gi...@apache.org>.
ztelur commented on a change in pull request #300:
URL: https://github.com/apache/dubbo-go-pixiu/pull/300#discussion_r752124963
##########
File path: pkg/client/mq/facade.go
##########
@@ -0,0 +1,92 @@
+/*
+ * 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 mq
+
+import (
+ "context"
+ "strings"
+)
+
+type ConsumerFacade interface {
+ // Subscribe message with specified broker and Topic, then handle msg with handler which send msg to real consumers
+ Subscribe(ctx context.Context, option ...Option) error
+ UnSubscribe(opts ...Option) error
+ Stop()
+}
+
+func GetConsumerManagerKey(topic []string, consumerGroup string) string {
+ return strings.Join((topic), ".") + "_" + consumerGroup
+}
+
+// MQOptions Consumer options
+// TODO: Add rocketmq params
+type MQOptions struct {
+ TopicList []string
+ ConsumeUrl string
+ CheckUrl string
+ ConsumerGroup string
+}
+
+func (o *MQOptions) ApplyOpts(opts ...Option) {
+ for _, opt := range opts {
+ opt(o)
+ }
+}
+
+func DefaultOptions() *MQOptions {
+ return &MQOptions{
+ TopicList: []string{"demo-topic"},
Review comment:
demo-topic ?
##########
File path: pkg/client/mq/kafka_facade.go
##########
@@ -0,0 +1,256 @@
+/*
+ * 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 mq
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "fmt"
+ "net/http"
+ "strconv"
+ "sync"
+ "time"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/logger"
+)
+
+import (
+ "github.com/Shopify/sarama"
+ perrors "github.com/pkg/errors"
+)
+
+type kafkaErrors struct {
+ count int
+ err string
+}
+
+func (ke kafkaErrors) Error() string {
+ return fmt.Sprintf("Failed to deliver %d messages due to %s", ke.count, ke.err)
+}
+
+func NewKafkaConsumerFacade(config KafkaConsumerConfig, consumerGroup string) (*KafkaConsumerFacade, error) {
+ c := sarama.NewConfig()
+ c.ClientID = config.ClientID
+ c.Metadata.Full = config.Metadata.Full
+ c.Metadata.Retry.Max = config.Metadata.Retry.Max
+ c.Metadata.Retry.Backoff = config.Metadata.Retry.Backoff
+ if config.ProtocolVersion != "" {
+ version, err := sarama.ParseKafkaVersion(config.ProtocolVersion)
+ if err != nil {
+ return nil, err
+ }
+ c.Version = version
+ }
+ client, err := sarama.NewConsumerGroup(config.Brokers, consumerGroup, c)
+ if err != nil {
+ return nil, err
+ }
+
+ return &KafkaConsumerFacade{consumerGroup: client, httpClient: &http.Client{Timeout: 5 * time.Second}, done: make(chan struct{})}, nil
+}
+
+type KafkaConsumerFacade struct {
+ consumerGroup sarama.ConsumerGroup
+ consumerManager map[string]func()
+ httpClient *http.Client
+ wg sync.WaitGroup
+ done chan struct{}
+}
+
+func (f *KafkaConsumerFacade) Subscribe(ctx context.Context, opts ...Option) error {
+ cOpt := DefaultOptions()
+ cOpt.ApplyOpts(opts...)
+ c, cancel := context.WithCancel(ctx)
+ key := GetConsumerManagerKey(cOpt.TopicList, cOpt.ConsumerGroup)
Review comment:
opts是可选的，里边有直接使用了依赖特定 Option 生成的 TopicList 和 ConsumerGroup，是否将必选字段直接作为函数参数？下边的Send函数也是
##########
File path: pkg/client/mq/kafka_facade.go
##########
@@ -0,0 +1,256 @@
+/*
+ * 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 mq
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "fmt"
+ "net/http"
+ "strconv"
+ "sync"
+ "time"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/logger"
+)
+
+import (
+ "github.com/Shopify/sarama"
+ perrors "github.com/pkg/errors"
+)
+
+type kafkaErrors struct {
+ count int
+ err string
+}
+
+func (ke kafkaErrors) Error() string {
+ return fmt.Sprintf("Failed to deliver %d messages due to %s", ke.count, ke.err)
+}
+
+func NewKafkaConsumerFacade(config KafkaConsumerConfig, consumerGroup string) (*KafkaConsumerFacade, error) {
+ c := sarama.NewConfig()
+ c.ClientID = config.ClientID
+ c.Metadata.Full = config.Metadata.Full
+ c.Metadata.Retry.Max = config.Metadata.Retry.Max
+ c.Metadata.Retry.Backoff = config.Metadata.Retry.Backoff
+ if config.ProtocolVersion != "" {
+ version, err := sarama.ParseKafkaVersion(config.ProtocolVersion)
+ if err != nil {
+ return nil, err
+ }
+ c.Version = version
+ }
+ client, err := sarama.NewConsumerGroup(config.Brokers, consumerGroup, c)
+ if err != nil {
+ return nil, err
+ }
+
+ return &KafkaConsumerFacade{consumerGroup: client, httpClient: &http.Client{Timeout: 5 * time.Second}, done: make(chan struct{})}, nil
+}
+
+type KafkaConsumerFacade struct {
+ consumerGroup sarama.ConsumerGroup
+ consumerManager map[string]func()
+ httpClient *http.Client
+ wg sync.WaitGroup
+ done chan struct{}
+}
+
+func (f *KafkaConsumerFacade) Subscribe(ctx context.Context, opts ...Option) error {
+ cOpt := DefaultOptions()
+ cOpt.ApplyOpts(opts...)
+ c, cancel := context.WithCancel(ctx)
+ key := GetConsumerManagerKey(cOpt.TopicList, cOpt.ConsumerGroup)
+ f.consumerManager[key] = cancel
+ f.wg.Add(2)
+ go f.consumeLoop(ctx, cOpt.TopicList, &consumerGroupHandler{cOpt.ConsumeUrl, f.httpClient})
+ go f.checkConsumerIsAlive(c, key, cOpt.CheckUrl)
+ return nil
+}
+
+func (f *KafkaConsumerFacade) consumeLoop(ctx context.Context, topics []string, handler sarama.ConsumerGroupHandler) {
+ for {
+ if _, ok := <-f.done; ok {
+ logger.Info("shutdown the consume loop")
+ break
+ }
+ if err := f.consumerGroup.Consume(ctx, topics, handler); err != nil {
+ logger.Warn("failed to consume the msg from kafka, %s", err.Error())
+ }
+ if ctx.Err() != nil {
+ // log consume stop
+ logger.Error("shutdown the consume loop due to %s", ctx.Err().Error())
+ break
+ }
+ }
+}
+
+type consumerGroupHandler struct {
+ consumerUrl string
+ httpClient *http.Client
+}
+
+func (c *consumerGroupHandler) Setup(session sarama.ConsumerGroupSession) error {
+ return nil
+}
+
+func (c *consumerGroupHandler) Cleanup(session sarama.ConsumerGroupSession) error {
+ return nil
+}
+
+func (c *consumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
+ for msg := range claim.Messages() {
+ session.MarkMessage(msg, "")
+ data, err := json.Marshal(MQMsgPush{Msg: []string{string(msg.Value)}})
+ if err != nil {
+ logger.Warn()
+ continue
+ }
+
+ req, err := http.NewRequest(http.MethodPost, c.consumerUrl, bytes.NewReader(data))
+ if err != nil {
+ logger.Warn()
+ continue
+ }
+ err = func() error {
+ resp, err := c.httpClient.Do(req)
+ if err != nil {
+ return err
+ }
+ defer resp.Body.Close()
+
+ if resp.StatusCode == http.StatusOK {
+ return nil
+ }
+ return perrors.New("failed send msg to consumer")
+ }()
+ if err != nil {
+ logger.Warn(err.Error())
+ }
+ }
+ return nil
+}
+
+// checkConsumerIsAlive make sure consumer is alive or would be removed from consumer list
+func (f *KafkaConsumerFacade) checkConsumerIsAlive(ctx context.Context, key string, checkUrl string) {
+ defer f.wg.Done()
+
+ ticker := time.NewTicker(15 * time.Second)
+ for {
+ select {
+ case <-f.done:
+ logger.Info()
+ case <-ticker.C:
+ lastCheck := 0
+ for i := 0; i < 5; i++ {
+ err := func() error {
+ req, err := http.NewRequest(http.MethodGet, checkUrl, bytes.NewReader([]byte{}))
+ if err != nil {
+ logger.Warn()
+ }
+
+ resp, err := f.httpClient.Do(req)
+ if err != nil {
+ logger.Warn()
+ }
+ defer resp.Body.Close()
+
+ lastCheck = resp.StatusCode
+ if resp.StatusCode != http.StatusOK {
+ return perrors.New("failed check consumer alive or not with status code " + strconv.Itoa(resp.StatusCode))
+ }
+
+ logger.Warn()
+ return nil
+ }()
+ if err != nil {
+ logger.Warn()
+ time.Sleep(10 * time.Millisecond)
+ } else {
+ break
+ }
+ }
+
+ if lastCheck != http.StatusOK {
+ f.consumerManager[key]()
+ delete(f.consumerManager, key)
+ }
+
+ }
+ }
+}
+
+func (f *KafkaConsumerFacade) UnSubscribe(opts ...Option) error {
+ return nil
+}
+
+func (f *KafkaConsumerFacade) Stop() {
+ close(f.done)
+ f.wg.Wait()
+}
+
+func NewKafkaProviderFacade(config KafkaProducerConfig) (*KafkaProducerFacade, error) {
+ c := sarama.NewConfig()
+ c.Producer.Return.Successes = true
+ c.Producer.Return.Errors = true
+ c.Producer.RequiredAcks = sarama.WaitForLocal
+ c.Metadata.Full = config.Metadata.Full
+ c.Metadata.Retry.Max = config.Metadata.Retry.Max
+ c.Metadata.Retry.Backoff = config.Metadata.Retry.Backoff
+ c.Producer.MaxMessageBytes = config.Producer.MaxMessageBytes
+ if config.ProtocolVersion != "" {
+ version, err := sarama.ParseKafkaVersion(config.ProtocolVersion)
+ if err != nil {
+ return nil, err
+ }
+ c.Version = version
+ }
+ producer, err := sarama.NewSyncProducer(config.Brokers, c)
+ if err != nil {
+ return nil, err
+ }
+ return &KafkaProducerFacade{producer: producer}, nil
+}
+
+type KafkaProducerFacade struct {
+ producer sarama.SyncProducer
+}
+
+func (k *KafkaProducerFacade) Send(msgs []string, opts ...Option) error {
+ pOpt := DefaultOptions()
+ pOpt.ApplyOpts(opts...)
+
+ pMsgs := make([]*sarama.ProducerMessage, 0)
Review comment:
直接声明和msg大小一样的切片？
##########
File path: pkg/client/mq/mq.go
##########
@@ -0,0 +1,156 @@
+/*
+ * 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 mq
+
+import (
+ "context"
+ "encoding/json"
+ "io/ioutil"
+ "strings"
+ "sync"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/client"
+ "github.com/apache/dubbo-go-pixiu/pkg/common/constant"
+ "github.com/apache/dubbo-go-pixiu/pkg/logger"
+)
+
+import (
+ perrors "github.com/pkg/errors"
+)
+
+var (
+ mqClient *Client
+ once sync.Once
+ consumerFacadeMap sync.Map
+)
+
+func NewSingletonMQClient(config Config) *Client {
+ if mqClient == nil {
+ once.Do(func() {
+ var err error
+ mqClient, err = NewMQClient(config)
+ if err != nil {
+ logger.Errorf("create mq client failed, %s", err.Error())
+ }
+ })
+ }
+ return mqClient
+}
+
+func NewMQClient(config Config) (*Client, error) {
+ var c *Client
+ ctx := context.Background()
+ switch config.MqType {
+ case constant.MQTypeKafka:
+ pf, err := NewKafkaProviderFacade(config.KafkaProducerConfig)
+ if err != nil {
+ return nil, err
+ }
+ c = &Client{
+ ctx: ctx,
+ producerFacade: pf,
+ kafkaConsumerConfig: config.KafkaConsumerConfig,
+ }
+ case constant.MQTypeRocketMQ:
+ return nil, perrors.New("rocketmq not support")
+ }
+
+ return c, nil
+}
+
+type Client struct {
+ ctx context.Context
+ producerFacade ProducerFacade
+ kafkaConsumerConfig KafkaConsumerConfig
+}
+
+func (c Client) Apply() error {
+ panic("implement me")
+}
+
+func (c Client) Close() error {
+ return nil
+}
+
+func (c Client) Call(req *client.Request) (res interface{}, err error) {
+ body, err := ioutil.ReadAll(req.IngressRequest.Body)
+ if err != nil {
+ return nil, err
+ }
+
+ paths := strings.Split(req.API.Path, "/")
+ if len(paths) < 3 {
+ return nil, perrors.New("failed to send message, broker or Topic not found")
+ }
+
+ switch MQActionStrToInt[paths[0]] {
+ case MQActionPublish:
+ var pReq MQProduceRequest
+ err = json.Unmarshal(body, &pReq)
+ if err != nil {
+ return nil, err
+ }
+ err = c.producerFacade.Send(pReq.Msg, WithTopic(pReq.Topic))
+ if err != nil {
+ return nil, err
+ }
+ case MQActionSubscribe:
+ var cReq MQSubscribeRequest
+ err = json.Unmarshal(body, &cReq)
+ if err != nil {
+ return nil, err
+ }
+ if _, ok := consumerFacadeMap.Load(cReq.ConsumerGroup); !ok {
Review comment:
Load有数据的分支是不是遗漏了
##########
File path: pkg/client/mq/mq.go
##########
@@ -0,0 +1,156 @@
+/*
+ * 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 mq
+
+import (
+ "context"
+ "encoding/json"
+ "io/ioutil"
+ "strings"
+ "sync"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/client"
+ "github.com/apache/dubbo-go-pixiu/pkg/common/constant"
+ "github.com/apache/dubbo-go-pixiu/pkg/logger"
+)
+
+import (
+ perrors "github.com/pkg/errors"
+)
+
+var (
+ mqClient *Client
+ once sync.Once
+ consumerFacadeMap sync.Map
+)
+
+func NewSingletonMQClient(config Config) *Client {
+ if mqClient == nil {
+ once.Do(func() {
+ var err error
+ mqClient, err = NewMQClient(config)
+ if err != nil {
+ logger.Errorf("create mq client failed, %s", err.Error())
+ }
+ })
+ }
+ return mqClient
+}
+
+func NewMQClient(config Config) (*Client, error) {
+ var c *Client
+ ctx := context.Background()
+ switch config.MqType {
+ case constant.MQTypeKafka:
+ pf, err := NewKafkaProviderFacade(config.KafkaProducerConfig)
+ if err != nil {
+ return nil, err
+ }
+ c = &Client{
+ ctx: ctx,
+ producerFacade: pf,
+ kafkaConsumerConfig: config.KafkaConsumerConfig,
+ }
+ case constant.MQTypeRocketMQ:
+ return nil, perrors.New("rocketmq not support")
+ }
+
+ return c, nil
+}
+
+type Client struct {
+ ctx context.Context
+ producerFacade ProducerFacade
+ kafkaConsumerConfig KafkaConsumerConfig
+}
+
+func (c Client) Apply() error {
+ panic("implement me")
+}
+
+func (c Client) Close() error {
+ return nil
+}
+
+func (c Client) Call(req *client.Request) (res interface{}, err error) {
+ body, err := ioutil.ReadAll(req.IngressRequest.Body)
+ if err != nil {
+ return nil, err
+ }
+
+ paths := strings.Split(req.API.Path, "/")
+ if len(paths) < 3 {
+ return nil, perrors.New("failed to send message, broker or Topic not found")
+ }
+
+ switch MQActionStrToInt[paths[0]] {
+ case MQActionPublish:
+ var pReq MQProduceRequest
+ err = json.Unmarshal(body, &pReq)
+ if err != nil {
+ return nil, err
+ }
+ err = c.producerFacade.Send(pReq.Msg, WithTopic(pReq.Topic))
+ if err != nil {
+ return nil, err
+ }
+ case MQActionSubscribe:
+ var cReq MQSubscribeRequest
+ err = json.Unmarshal(body, &cReq)
+ if err != nil {
+ return nil, err
+ }
+ if _, ok := consumerFacadeMap.Load(cReq.ConsumerGroup); !ok {
+ facade, err := NewKafkaConsumerFacade(c.kafkaConsumerConfig, cReq.ConsumerGroup)
+ if err != nil {
+ return nil, err
+ }
+ consumerFacadeMap.Store(cReq.ConsumerGroup, facade)
+ if f, ok := consumerFacadeMap.Load(cReq.ConsumerGroup); ok {
Review comment:
为什么不直接使用 facade 变量？
##########
File path: pkg/client/mq/kafka_facade.go
##########
@@ -0,0 +1,256 @@
+/*
+ * 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 mq
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "fmt"
+ "net/http"
+ "strconv"
+ "sync"
+ "time"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/logger"
+)
+
+import (
+ "github.com/Shopify/sarama"
+ perrors "github.com/pkg/errors"
+)
+
+type kafkaErrors struct {
+ count int
+ err string
+}
+
+func (ke kafkaErrors) Error() string {
+ return fmt.Sprintf("Failed to deliver %d messages due to %s", ke.count, ke.err)
+}
+
+func NewKafkaConsumerFacade(config KafkaConsumerConfig, consumerGroup string) (*KafkaConsumerFacade, error) {
+ c := sarama.NewConfig()
+ c.ClientID = config.ClientID
+ c.Metadata.Full = config.Metadata.Full
+ c.Metadata.Retry.Max = config.Metadata.Retry.Max
+ c.Metadata.Retry.Backoff = config.Metadata.Retry.Backoff
+ if config.ProtocolVersion != "" {
+ version, err := sarama.ParseKafkaVersion(config.ProtocolVersion)
+ if err != nil {
+ return nil, err
+ }
+ c.Version = version
+ }
+ client, err := sarama.NewConsumerGroup(config.Brokers, consumerGroup, c)
+ if err != nil {
+ return nil, err
+ }
+
+ return &KafkaConsumerFacade{consumerGroup: client, httpClient: &http.Client{Timeout: 5 * time.Second}, done: make(chan struct{})}, nil
+}
+
+type KafkaConsumerFacade struct {
+ consumerGroup sarama.ConsumerGroup
+ consumerManager map[string]func()
+ httpClient *http.Client
+ wg sync.WaitGroup
+ done chan struct{}
+}
+
+func (f *KafkaConsumerFacade) Subscribe(ctx context.Context, opts ...Option) error {
+ cOpt := DefaultOptions()
+ cOpt.ApplyOpts(opts...)
+ c, cancel := context.WithCancel(ctx)
+ key := GetConsumerManagerKey(cOpt.TopicList, cOpt.ConsumerGroup)
+ f.consumerManager[key] = cancel
+ f.wg.Add(2)
+ go f.consumeLoop(ctx, cOpt.TopicList, &consumerGroupHandler{cOpt.ConsumeUrl, f.httpClient})
+ go f.checkConsumerIsAlive(c, key, cOpt.CheckUrl)
+ return nil
+}
+
+func (f *KafkaConsumerFacade) consumeLoop(ctx context.Context, topics []string, handler sarama.ConsumerGroupHandler) {
+ for {
+ if _, ok := <-f.done; ok {
+ logger.Info("shutdown the consume loop")
+ break
+ }
+ if err := f.consumerGroup.Consume(ctx, topics, handler); err != nil {
+ logger.Warn("failed to consume the msg from kafka, %s", err.Error())
+ }
+ if ctx.Err() != nil {
+ // log consume stop
+ logger.Error("shutdown the consume loop due to %s", ctx.Err().Error())
+ break
+ }
+ }
+}
+
+type consumerGroupHandler struct {
+ consumerUrl string
+ httpClient *http.Client
+}
+
+func (c *consumerGroupHandler) Setup(session sarama.ConsumerGroupSession) error {
+ return nil
+}
+
+func (c *consumerGroupHandler) Cleanup(session sarama.ConsumerGroupSession) error {
+ return nil
+}
+
+func (c *consumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
+ for msg := range claim.Messages() {
+ session.MarkMessage(msg, "")
+ data, err := json.Marshal(MQMsgPush{Msg: []string{string(msg.Value)}})
+ if err != nil {
+ logger.Warn()
+ continue
+ }
+
+ req, err := http.NewRequest(http.MethodPost, c.consumerUrl, bytes.NewReader(data))
+ if err != nil {
+ logger.Warn()
+ continue
+ }
+ err = func() error {
+ resp, err := c.httpClient.Do(req)
+ if err != nil {
+ return err
+ }
+ defer resp.Body.Close()
+
+ if resp.StatusCode == http.StatusOK {
+ return nil
+ }
+ return perrors.New("failed send msg to consumer")
+ }()
+ if err != nil {
+ logger.Warn(err.Error())
+ }
+ }
+ return nil
+}
+
+// checkConsumerIsAlive make sure consumer is alive or would be removed from consumer list
+func (f *KafkaConsumerFacade) checkConsumerIsAlive(ctx context.Context, key string, checkUrl string) {
+ defer f.wg.Done()
+
+ ticker := time.NewTicker(15 * time.Second)
Review comment:
函数退出是否需要关闭 ticker
##########
File path: pkg/client/mq/kafka_facade.go
##########
@@ -0,0 +1,256 @@
+/*
+ * 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 mq
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "fmt"
+ "net/http"
+ "strconv"
+ "sync"
+ "time"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/logger"
+)
+
+import (
+ "github.com/Shopify/sarama"
+ perrors "github.com/pkg/errors"
+)
+
+type kafkaErrors struct {
+ count int
+ err string
+}
+
+func (ke kafkaErrors) Error() string {
+ return fmt.Sprintf("Failed to deliver %d messages due to %s", ke.count, ke.err)
+}
+
+func NewKafkaConsumerFacade(config KafkaConsumerConfig, consumerGroup string) (*KafkaConsumerFacade, error) {
+ c := sarama.NewConfig()
+ c.ClientID = config.ClientID
+ c.Metadata.Full = config.Metadata.Full
+ c.Metadata.Retry.Max = config.Metadata.Retry.Max
+ c.Metadata.Retry.Backoff = config.Metadata.Retry.Backoff
+ if config.ProtocolVersion != "" {
+ version, err := sarama.ParseKafkaVersion(config.ProtocolVersion)
+ if err != nil {
+ return nil, err
+ }
+ c.Version = version
+ }
+ client, err := sarama.NewConsumerGroup(config.Brokers, consumerGroup, c)
+ if err != nil {
+ return nil, err
+ }
+
+ return &KafkaConsumerFacade{consumerGroup: client, httpClient: &http.Client{Timeout: 5 * time.Second}, done: make(chan struct{})}, nil
+}
+
+type KafkaConsumerFacade struct {
+ consumerGroup sarama.ConsumerGroup
+ consumerManager map[string]func()
+ httpClient *http.Client
+ wg sync.WaitGroup
+ done chan struct{}
+}
+
+func (f *KafkaConsumerFacade) Subscribe(ctx context.Context, opts ...Option) error {
+ cOpt := DefaultOptions()
+ cOpt.ApplyOpts(opts...)
+ c, cancel := context.WithCancel(ctx)
+ key := GetConsumerManagerKey(cOpt.TopicList, cOpt.ConsumerGroup)
+ f.consumerManager[key] = cancel
+ f.wg.Add(2)
+ go f.consumeLoop(ctx, cOpt.TopicList, &consumerGroupHandler{cOpt.ConsumeUrl, f.httpClient})
+ go f.checkConsumerIsAlive(c, key, cOpt.CheckUrl)
+ return nil
+}
+
+func (f *KafkaConsumerFacade) consumeLoop(ctx context.Context, topics []string, handler sarama.ConsumerGroupHandler) {
+ for {
+ if _, ok := <-f.done; ok {
Review comment:
是不是会被阻塞，用select模式？
##########
File path: pkg/client/mq/mq.go
##########
@@ -0,0 +1,156 @@
+/*
+ * 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 mq
+
+import (
+ "context"
+ "encoding/json"
+ "io/ioutil"
+ "strings"
+ "sync"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/client"
+ "github.com/apache/dubbo-go-pixiu/pkg/common/constant"
+ "github.com/apache/dubbo-go-pixiu/pkg/logger"
+)
+
+import (
+ perrors "github.com/pkg/errors"
+)
+
+var (
+ mqClient *Client
+ once sync.Once
+ consumerFacadeMap sync.Map
+)
+
+func NewSingletonMQClient(config Config) *Client {
+ if mqClient == nil {
+ once.Do(func() {
+ var err error
+ mqClient, err = NewMQClient(config)
+ if err != nil {
+ logger.Errorf("create mq client failed, %s", err.Error())
+ }
+ })
+ }
+ return mqClient
+}
+
+func NewMQClient(config Config) (*Client, error) {
+ var c *Client
+ ctx := context.Background()
+ switch config.MqType {
+ case constant.MQTypeKafka:
+ pf, err := NewKafkaProviderFacade(config.KafkaProducerConfig)
+ if err != nil {
+ return nil, err
+ }
+ c = &Client{
+ ctx: ctx,
+ producerFacade: pf,
+ kafkaConsumerConfig: config.KafkaConsumerConfig,
+ }
+ case constant.MQTypeRocketMQ:
+ return nil, perrors.New("rocketmq not support")
+ }
+
+ return c, nil
+}
+
+type Client struct {
+ ctx context.Context
+ producerFacade ProducerFacade
+ kafkaConsumerConfig KafkaConsumerConfig
+}
+
+func (c Client) Apply() error {
+ panic("implement me")
+}
+
+func (c Client) Close() error {
+ return nil
+}
+
+func (c Client) Call(req *client.Request) (res interface{}, err error) {
+ body, err := ioutil.ReadAll(req.IngressRequest.Body)
+ if err != nil {
+ return nil, err
+ }
+
+ paths := strings.Split(req.API.Path, "/")
+ if len(paths) < 3 {
+ return nil, perrors.New("failed to send message, broker or Topic not found")
+ }
+
+ switch MQActionStrToInt[paths[0]] {
+ case MQActionPublish:
+ var pReq MQProduceRequest
+ err = json.Unmarshal(body, &pReq)
+ if err != nil {
+ return nil, err
+ }
+ err = c.producerFacade.Send(pReq.Msg, WithTopic(pReq.Topic))
+ if err != nil {
+ return nil, err
+ }
+ case MQActionSubscribe:
+ var cReq MQSubscribeRequest
+ err = json.Unmarshal(body, &cReq)
+ if err != nil {
+ return nil, err
+ }
+ if _, ok := consumerFacadeMap.Load(cReq.ConsumerGroup); !ok {
Review comment:
Load有数据时的分支是不是漏掉了？
##########
File path: pkg/client/mq/facade.go
##########
@@ -0,0 +1,92 @@
+/*
+ * 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 mq
+
+import (
+ "context"
+ "strings"
+)
+
+type ConsumerFacade interface {
+ // Subscribe message with specified broker and Topic, then handle msg with handler which send msg to real consumers
+ Subscribe(ctx context.Context, option ...Option) error
+ UnSubscribe(opts ...Option) error
+ Stop()
+}
+
+func GetConsumerManagerKey(topic []string, consumerGroup string) string {
+ return strings.Join((topic), ".") + "_" + consumerGroup
+}
+
+// MQOptions Consumer options
+// TODO: Add rocketmq params
+type MQOptions struct {
Review comment:
配置类 MQOptions 也有自己的 functional Options 是不是有些过度设计
##########
File path: pkg/client/mq/kafka_facade.go
##########
@@ -0,0 +1,256 @@
+/*
+ * 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 mq
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "fmt"
+ "net/http"
+ "strconv"
+ "sync"
+ "time"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/logger"
+)
+
+import (
+ "github.com/Shopify/sarama"
+ perrors "github.com/pkg/errors"
+)
+
+type kafkaErrors struct {
+ count int
+ err string
+}
+
+func (ke kafkaErrors) Error() string {
+ return fmt.Sprintf("Failed to deliver %d messages due to %s", ke.count, ke.err)
+}
+
+func NewKafkaConsumerFacade(config KafkaConsumerConfig, consumerGroup string) (*KafkaConsumerFacade, error) {
+ c := sarama.NewConfig()
+ c.ClientID = config.ClientID
+ c.Metadata.Full = config.Metadata.Full
+ c.Metadata.Retry.Max = config.Metadata.Retry.Max
+ c.Metadata.Retry.Backoff = config.Metadata.Retry.Backoff
+ if config.ProtocolVersion != "" {
+ version, err := sarama.ParseKafkaVersion(config.ProtocolVersion)
+ if err != nil {
+ return nil, err
+ }
+ c.Version = version
+ }
+ client, err := sarama.NewConsumerGroup(config.Brokers, consumerGroup, c)
+ if err != nil {
+ return nil, err
+ }
+
+ return &KafkaConsumerFacade{consumerGroup: client, httpClient: &http.Client{Timeout: 5 * time.Second}, done: make(chan struct{})}, nil
+}
+
+type KafkaConsumerFacade struct {
+ consumerGroup sarama.ConsumerGroup
+ consumerManager map[string]func()
+ httpClient *http.Client
+ wg sync.WaitGroup
+ done chan struct{}
+}
+
+func (f *KafkaConsumerFacade) Subscribe(ctx context.Context, opts ...Option) error {
+ cOpt := DefaultOptions()
+ cOpt.ApplyOpts(opts...)
+ c, cancel := context.WithCancel(ctx)
+ key := GetConsumerManagerKey(cOpt.TopicList, cOpt.ConsumerGroup)
+ f.consumerManager[key] = cancel
+ f.wg.Add(2)
+ go f.consumeLoop(ctx, cOpt.TopicList, &consumerGroupHandler{cOpt.ConsumeUrl, f.httpClient})
+ go f.checkConsumerIsAlive(c, key, cOpt.CheckUrl)
+ return nil
+}
+
+func (f *KafkaConsumerFacade) consumeLoop(ctx context.Context, topics []string, handler sarama.ConsumerGroupHandler) {
+ for {
+ if _, ok := <-f.done; ok {
+ logger.Info("shutdown the consume loop")
+ break
+ }
+ if err := f.consumerGroup.Consume(ctx, topics, handler); err != nil {
+ logger.Warn("failed to consume the msg from kafka, %s", err.Error())
+ }
+ if ctx.Err() != nil {
+ // log consume stop
+ logger.Error("shutdown the consume loop due to %s", ctx.Err().Error())
+ break
+ }
+ }
+}
+
+type consumerGroupHandler struct {
+ consumerUrl string
+ httpClient *http.Client
+}
+
+func (c *consumerGroupHandler) Setup(session sarama.ConsumerGroupSession) error {
+ return nil
+}
+
+func (c *consumerGroupHandler) Cleanup(session sarama.ConsumerGroupSession) error {
+ return nil
+}
+
+func (c *consumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
+ for msg := range claim.Messages() {
+ session.MarkMessage(msg, "")
+ data, err := json.Marshal(MQMsgPush{Msg: []string{string(msg.Value)}})
+ if err != nil {
+ logger.Warn()
+ continue
+ }
+
+ req, err := http.NewRequest(http.MethodPost, c.consumerUrl, bytes.NewReader(data))
+ if err != nil {
+ logger.Warn()
+ continue
+ }
+ err = func() error {
+ resp, err := c.httpClient.Do(req)
+ if err != nil {
+ return err
+ }
+ defer resp.Body.Close()
+
+ if resp.StatusCode == http.StatusOK {
+ return nil
+ }
+ return perrors.New("failed send msg to consumer")
+ }()
+ if err != nil {
+ logger.Warn(err.Error())
+ }
+ }
+ return nil
+}
+
+// checkConsumerIsAlive make sure consumer is alive or would be removed from consumer list
+func (f *KafkaConsumerFacade) checkConsumerIsAlive(ctx context.Context, key string, checkUrl string) {
+ defer f.wg.Done()
+
+ ticker := time.NewTicker(15 * time.Second)
+ for {
+ select {
+ case <-f.done:
+ logger.Info()
Review comment:
需要打印明确日志，还有几处可以一起修改
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go-pixiu] codecov-commenter edited a comment on pull request #300: [feature eventmesh]eventmesh
Posted by GitBox <gi...@apache.org>.
codecov-commenter edited a comment on pull request #300:
URL: https://github.com/apache/dubbo-go-pixiu/pull/300#issuecomment-968298380
# [Codecov](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#300](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (b97b380) into [develop](https://codecov.io/gh/apache/dubbo-go-pixiu/commit/d1930df3e910c5f948380c703fbbce4f49490d2d?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (d1930df) will **not change** coverage.
> The diff coverage is `n/a`.
[](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## develop #300 +/- ##
========================================
Coverage 33.93% 33.93%
========================================
Files 61 61
Lines 3165 3165
========================================
Hits 1074 1074
Misses 1979 1979
Partials 112 112
```
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [d1930df...b97b380](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go-pixiu] AlexStocks commented on a change in pull request #300: [feature eventmesh]eventmesh
Posted by GitBox <gi...@apache.org>.
AlexStocks commented on a change in pull request #300:
URL: https://github.com/apache/dubbo-go-pixiu/pull/300#discussion_r749007345
##########
File path: pkg/client/mq/kafka_facade.go
##########
@@ -0,0 +1,237 @@
+/*
+ * 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 mq
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "net/http"
+ "strconv"
+ "sync"
+ "time"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/filter/event"
+ "github.com/apache/dubbo-go-pixiu/pkg/logger"
+)
+
+import (
+ "github.com/Shopify/sarama"
+ perrors "github.com/pkg/errors"
+)
+
+func NewKafkaConsumerFacade(config event.KafkaConsumerConfig) (*KafkaConsumerFacade, error) {
+ c := sarama.NewConfig()
+ c.ClientID = config.ClientID
+ c.Metadata.Full = config.Metadata.Full
+ c.Metadata.Retry.Max = config.Metadata.Retry.Max
+ c.Metadata.Retry.Backoff = config.Metadata.Retry.Backoff
+ if config.ProtocolVersion != "" {
+ version, err := sarama.ParseKafkaVersion(config.ProtocolVersion)
+ if err != nil {
+ return nil, err
+ }
+ c.Version = version
+ }
+ consumer, err := sarama.NewConsumer(config.Brokers, c)
+ if err != nil {
+ return nil, err
+ }
+
+ return &KafkaConsumerFacade{consumer: consumer, httpClient: &http.Client{Timeout: 5 * time.Second}, done: make(chan struct{})}, nil
+}
+
+type KafkaConsumerFacade struct {
+ consumer sarama.Consumer
+ consumerManager map[string]func()
+ rwLock sync.RWMutex
+ httpClient *http.Client
+ wg sync.WaitGroup
+ done chan struct{}
+}
+
+func (f *KafkaConsumerFacade) Subscribe(ctx context.Context, opts ...Option) error {
+ cOpt := DefaultOptions()
+ cOpt.ApplyOpts(opts...)
+ partConsumer, err := f.consumer.ConsumePartition(cOpt.Topic, int32(cOpt.Partition), sarama.OffsetOldest)
+ if err != nil {
+ return err
+ }
+ c, cancel := context.WithCancel(ctx)
+ key := GetConsumerManagerKey(cOpt.Topic, int32(cOpt.Partition))
+ f.rwLock.Lock()
+ defer f.rwLock.Unlock()
+ f.consumerManager[key] = cancel
+ f.wg.Add(2)
+ go f.ConsumePartitions(c, partConsumer, cOpt.ConsumeUrl)
+ go f.checkConsumerIsAlive(c, key, cOpt.CheckUrl)
+ return nil
+}
+
+// ConsumePartitions consume function
+func (f *KafkaConsumerFacade) ConsumePartitions(ctx context.Context, partConsumer sarama.PartitionConsumer, consumeUrl string) {
+ defer f.wg.Done()
+
+ for {
+ select {
+ case <-f.done:
+ logger.Info()
Review comment:
no log?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go-pixiu] codecov-commenter edited a comment on pull request #300: [feature eventmesh]eventmesh
Posted by GitBox <gi...@apache.org>.
codecov-commenter edited a comment on pull request #300:
URL: https://github.com/apache/dubbo-go-pixiu/pull/300#issuecomment-968298380
# [Codecov](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#300](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (42d5b86) into [develop](https://codecov.io/gh/apache/dubbo-go-pixiu/commit/ec68432b71b733a1d87b7a182ded2004fed073d9?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (ec68432) will **not change** coverage.
> The diff coverage is `n/a`.
[](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## develop #300 +/- ##
========================================
Coverage 33.98% 33.98%
========================================
Files 61 61
Lines 3160 3160
========================================
Hits 1074 1074
Misses 1974 1974
Partials 112 112
```
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [ec68432...42d5b86](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go-pixiu] tydhot commented on a change in pull request #300: [feature eventmesh]eventmesh
Posted by GitBox <gi...@apache.org>.
tydhot commented on a change in pull request #300:
URL: https://github.com/apache/dubbo-go-pixiu/pull/300#discussion_r753800307
##########
File path: pkg/client/mq/kafka_facade.go
##########
@@ -0,0 +1,256 @@
+/*
+ * 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 mq
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "fmt"
+ "net/http"
+ "strconv"
+ "sync"
+ "time"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/logger"
+)
+
+import (
+ "github.com/Shopify/sarama"
+ perrors "github.com/pkg/errors"
+)
+
+type kafkaErrors struct {
+ count int
+ err string
+}
+
+func (ke kafkaErrors) Error() string {
+ return fmt.Sprintf("Failed to deliver %d messages due to %s", ke.count, ke.err)
+}
+
+func NewKafkaConsumerFacade(config KafkaConsumerConfig, consumerGroup string) (*KafkaConsumerFacade, error) {
+ c := sarama.NewConfig()
+ c.ClientID = config.ClientID
+ c.Metadata.Full = config.Metadata.Full
+ c.Metadata.Retry.Max = config.Metadata.Retry.Max
+ c.Metadata.Retry.Backoff = config.Metadata.Retry.Backoff
+ if config.ProtocolVersion != "" {
+ version, err := sarama.ParseKafkaVersion(config.ProtocolVersion)
+ if err != nil {
+ return nil, err
+ }
+ c.Version = version
+ }
+ client, err := sarama.NewConsumerGroup(config.Brokers, consumerGroup, c)
+ if err != nil {
+ return nil, err
+ }
+
+ return &KafkaConsumerFacade{consumerGroup: client, httpClient: &http.Client{Timeout: 5 * time.Second}, done: make(chan struct{})}, nil
+}
+
+type KafkaConsumerFacade struct {
+ consumerGroup sarama.ConsumerGroup
+ consumerManager map[string]func()
+ httpClient *http.Client
+ wg sync.WaitGroup
+ done chan struct{}
+}
+
+func (f *KafkaConsumerFacade) Subscribe(ctx context.Context, opts ...Option) error {
+ cOpt := DefaultOptions()
+ cOpt.ApplyOpts(opts...)
+ c, cancel := context.WithCancel(ctx)
+ key := GetConsumerManagerKey(cOpt.TopicList, cOpt.ConsumerGroup)
+ f.consumerManager[key] = cancel
+ f.wg.Add(2)
+ go f.consumeLoop(ctx, cOpt.TopicList, &consumerGroupHandler{cOpt.ConsumeUrl, f.httpClient})
+ go f.checkConsumerIsAlive(c, key, cOpt.CheckUrl)
+ return nil
+}
+
+func (f *KafkaConsumerFacade) consumeLoop(ctx context.Context, topics []string, handler sarama.ConsumerGroupHandler) {
+ for {
+ if _, ok := <-f.done; ok {
Review comment:
之前是select， 是review dog建议改成这样
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go-pixiu] tydhot commented on a change in pull request #300: [feature eventmesh]eventmesh
Posted by GitBox <gi...@apache.org>.
tydhot commented on a change in pull request #300:
URL: https://github.com/apache/dubbo-go-pixiu/pull/300#discussion_r753804635
##########
File path: pkg/client/mq/kafka_facade.go
##########
@@ -0,0 +1,256 @@
+/*
+ * 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 mq
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "fmt"
+ "net/http"
+ "strconv"
+ "sync"
+ "time"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/logger"
+)
+
+import (
+ "github.com/Shopify/sarama"
+ perrors "github.com/pkg/errors"
+)
+
+type kafkaErrors struct {
+ count int
+ err string
+}
+
+func (ke kafkaErrors) Error() string {
+ return fmt.Sprintf("Failed to deliver %d messages due to %s", ke.count, ke.err)
+}
+
+func NewKafkaConsumerFacade(config KafkaConsumerConfig, consumerGroup string) (*KafkaConsumerFacade, error) {
+ c := sarama.NewConfig()
+ c.ClientID = config.ClientID
+ c.Metadata.Full = config.Metadata.Full
+ c.Metadata.Retry.Max = config.Metadata.Retry.Max
+ c.Metadata.Retry.Backoff = config.Metadata.Retry.Backoff
+ if config.ProtocolVersion != "" {
+ version, err := sarama.ParseKafkaVersion(config.ProtocolVersion)
+ if err != nil {
+ return nil, err
+ }
+ c.Version = version
+ }
+ client, err := sarama.NewConsumerGroup(config.Brokers, consumerGroup, c)
+ if err != nil {
+ return nil, err
+ }
+
+ return &KafkaConsumerFacade{consumerGroup: client, httpClient: &http.Client{Timeout: 5 * time.Second}, done: make(chan struct{})}, nil
+}
+
+type KafkaConsumerFacade struct {
+ consumerGroup sarama.ConsumerGroup
+ consumerManager map[string]func()
+ httpClient *http.Client
+ wg sync.WaitGroup
+ done chan struct{}
+}
+
+func (f *KafkaConsumerFacade) Subscribe(ctx context.Context, opts ...Option) error {
+ cOpt := DefaultOptions()
+ cOpt.ApplyOpts(opts...)
+ c, cancel := context.WithCancel(ctx)
+ key := GetConsumerManagerKey(cOpt.TopicList, cOpt.ConsumerGroup)
+ f.consumerManager[key] = cancel
+ f.wg.Add(2)
+ go f.consumeLoop(ctx, cOpt.TopicList, &consumerGroupHandler{cOpt.ConsumeUrl, f.httpClient})
+ go f.checkConsumerIsAlive(c, key, cOpt.CheckUrl)
+ return nil
+}
+
+func (f *KafkaConsumerFacade) consumeLoop(ctx context.Context, topics []string, handler sarama.ConsumerGroupHandler) {
+ for {
+ if _, ok := <-f.done; ok {
+ logger.Info("shutdown the consume loop")
+ break
+ }
+ if err := f.consumerGroup.Consume(ctx, topics, handler); err != nil {
+ logger.Warn("failed to consume the msg from kafka, %s", err.Error())
+ }
+ if ctx.Err() != nil {
+ // log consume stop
+ logger.Error("shutdown the consume loop due to %s", ctx.Err().Error())
+ break
+ }
+ }
+}
+
+type consumerGroupHandler struct {
+ consumerUrl string
+ httpClient *http.Client
+}
+
+func (c *consumerGroupHandler) Setup(session sarama.ConsumerGroupSession) error {
+ return nil
+}
+
+func (c *consumerGroupHandler) Cleanup(session sarama.ConsumerGroupSession) error {
+ return nil
+}
+
+func (c *consumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
+ for msg := range claim.Messages() {
+ session.MarkMessage(msg, "")
+ data, err := json.Marshal(MQMsgPush{Msg: []string{string(msg.Value)}})
+ if err != nil {
+ logger.Warn()
+ continue
+ }
+
+ req, err := http.NewRequest(http.MethodPost, c.consumerUrl, bytes.NewReader(data))
+ if err != nil {
+ logger.Warn()
+ continue
+ }
+ err = func() error {
+ resp, err := c.httpClient.Do(req)
+ if err != nil {
+ return err
+ }
+ defer resp.Body.Close()
+
+ if resp.StatusCode == http.StatusOK {
+ return nil
+ }
+ return perrors.New("failed send msg to consumer")
+ }()
+ if err != nil {
+ logger.Warn(err.Error())
+ }
+ }
+ return nil
+}
+
+// checkConsumerIsAlive make sure consumer is alive or would be removed from consumer list
+func (f *KafkaConsumerFacade) checkConsumerIsAlive(ctx context.Context, key string, checkUrl string) {
+ defer f.wg.Done()
+
+ ticker := time.NewTicker(15 * time.Second)
Review comment:
ok
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go-pixiu] ztelur commented on a change in pull request #300: [feature eventmesh]eventmesh
Posted by GitBox <gi...@apache.org>.
ztelur commented on a change in pull request #300:
URL: https://github.com/apache/dubbo-go-pixiu/pull/300#discussion_r754063536
##########
File path: pkg/client/mq/mq.go
##########
@@ -0,0 +1,156 @@
+/*
+ * 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 mq
+
+import (
+ "context"
+ "encoding/json"
+ "io/ioutil"
+ "strings"
+ "sync"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/client"
+ "github.com/apache/dubbo-go-pixiu/pkg/common/constant"
+ "github.com/apache/dubbo-go-pixiu/pkg/logger"
+)
+
+import (
+ perrors "github.com/pkg/errors"
+)
+
+var (
+ mqClient *Client
+ once sync.Once
+ consumerFacadeMap sync.Map
+)
+
+func NewSingletonMQClient(config Config) *Client {
+ if mqClient == nil {
+ once.Do(func() {
+ var err error
+ mqClient, err = NewMQClient(config)
+ if err != nil {
+ logger.Errorf("create mq client failed, %s", err.Error())
+ }
+ })
+ }
+ return mqClient
+}
+
+func NewMQClient(config Config) (*Client, error) {
+ var c *Client
+ ctx := context.Background()
+ switch config.MqType {
+ case constant.MQTypeKafka:
+ pf, err := NewKafkaProviderFacade(config.KafkaProducerConfig)
+ if err != nil {
+ return nil, err
+ }
+ c = &Client{
+ ctx: ctx,
+ producerFacade: pf,
+ kafkaConsumerConfig: config.KafkaConsumerConfig,
+ }
+ case constant.MQTypeRocketMQ:
+ return nil, perrors.New("rocketmq not support")
+ }
+
+ return c, nil
+}
+
+type Client struct {
+ ctx context.Context
+ producerFacade ProducerFacade
+ kafkaConsumerConfig KafkaConsumerConfig
+}
+
+func (c Client) Apply() error {
+ panic("implement me")
+}
+
+func (c Client) Close() error {
+ return nil
+}
+
+func (c Client) Call(req *client.Request) (res interface{}, err error) {
+ body, err := ioutil.ReadAll(req.IngressRequest.Body)
+ if err != nil {
+ return nil, err
+ }
+
+ paths := strings.Split(req.API.Path, "/")
+ if len(paths) < 3 {
+ return nil, perrors.New("failed to send message, broker or Topic not found")
+ }
+
+ switch MQActionStrToInt[paths[0]] {
+ case MQActionPublish:
+ var pReq MQProduceRequest
+ err = json.Unmarshal(body, &pReq)
+ if err != nil {
+ return nil, err
+ }
+ err = c.producerFacade.Send(pReq.Msg, WithTopic(pReq.Topic))
+ if err != nil {
+ return nil, err
+ }
+ case MQActionSubscribe:
+ var cReq MQSubscribeRequest
+ err = json.Unmarshal(body, &cReq)
+ if err != nil {
+ return nil, err
+ }
+ if _, ok := consumerFacadeMap.Load(cReq.ConsumerGroup); !ok {
Review comment:
可能是我理解的有些问题， MQSubscribe中 ConsumerGroup 相同，但是 topicList 不同的场景是不是有些问题？
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go-pixiu] tydhot commented on a change in pull request #300: [feature eventmesh]eventmesh
Posted by GitBox <gi...@apache.org>.
tydhot commented on a change in pull request #300:
URL: https://github.com/apache/dubbo-go-pixiu/pull/300#discussion_r753801407
##########
File path: pkg/client/mq/facade.go
##########
@@ -0,0 +1,92 @@
+/*
+ * 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 mq
+
+import (
+ "context"
+ "strings"
+)
+
+type ConsumerFacade interface {
+ // Subscribe message with specified broker and Topic, then handle msg with handler which send msg to real consumers
+ Subscribe(ctx context.Context, option ...Option) error
+ UnSubscribe(opts ...Option) error
+ Stop()
+}
+
+func GetConsumerManagerKey(topic []string, consumerGroup string) string {
+ return strings.Join((topic), ".") + "_" + consumerGroup
+}
+
+// MQOptions Consumer options
+// TODO: Add rocketmq params
+type MQOptions struct {
Review comment:
这个我觉得暂时问题不大 后面再简化也不迟
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go-pixiu] codecov-commenter edited a comment on pull request #300: [feature eventmesh]eventmesh
Posted by GitBox <gi...@apache.org>.
codecov-commenter edited a comment on pull request #300:
URL: https://github.com/apache/dubbo-go-pixiu/pull/300#issuecomment-968298380
# [Codecov](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#300](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (db49d94) into [develop](https://codecov.io/gh/apache/dubbo-go-pixiu/commit/5bef4853016d8bfb0da4db715ccc5480cbcde7a6?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (5bef485) will **not change** coverage.
> The diff coverage is `n/a`.
[](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## develop #300 +/- ##
========================================
Coverage 33.93% 33.93%
========================================
Files 61 61
Lines 3165 3165
========================================
Hits 1074 1074
Misses 1979 1979
Partials 112 112
```
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [5bef485...db49d94](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go-pixiu] codecov-commenter edited a comment on pull request #300: [feature eventmesh]eventmesh
Posted by GitBox <gi...@apache.org>.
codecov-commenter edited a comment on pull request #300:
URL: https://github.com/apache/dubbo-go-pixiu/pull/300#issuecomment-968298380
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go-pixiu] tydhot commented on pull request #300: [feature eventmesh]eventmesh
Posted by GitBox <gi...@apache.org>.
tydhot commented on pull request #300:
URL: https://github.com/apache/dubbo-go-pixiu/pull/300#issuecomment-971544435
@ztelur 3q. I have fixed them
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go-pixiu] tydhot commented on a change in pull request #300: [feature eventmesh]eventmesh
Posted by GitBox <gi...@apache.org>.
tydhot commented on a change in pull request #300:
URL: https://github.com/apache/dubbo-go-pixiu/pull/300#discussion_r753800487
##########
File path: pkg/client/mq/mq.go
##########
@@ -0,0 +1,156 @@
+/*
+ * 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 mq
+
+import (
+ "context"
+ "encoding/json"
+ "io/ioutil"
+ "strings"
+ "sync"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/client"
+ "github.com/apache/dubbo-go-pixiu/pkg/common/constant"
+ "github.com/apache/dubbo-go-pixiu/pkg/logger"
+)
+
+import (
+ perrors "github.com/pkg/errors"
+)
+
+var (
+ mqClient *Client
+ once sync.Once
+ consumerFacadeMap sync.Map
+)
+
+func NewSingletonMQClient(config Config) *Client {
+ if mqClient == nil {
+ once.Do(func() {
+ var err error
+ mqClient, err = NewMQClient(config)
+ if err != nil {
+ logger.Errorf("create mq client failed, %s", err.Error())
+ }
+ })
+ }
+ return mqClient
+}
+
+func NewMQClient(config Config) (*Client, error) {
+ var c *Client
+ ctx := context.Background()
+ switch config.MqType {
+ case constant.MQTypeKafka:
+ pf, err := NewKafkaProviderFacade(config.KafkaProducerConfig)
+ if err != nil {
+ return nil, err
+ }
+ c = &Client{
+ ctx: ctx,
+ producerFacade: pf,
+ kafkaConsumerConfig: config.KafkaConsumerConfig,
+ }
+ case constant.MQTypeRocketMQ:
+ return nil, perrors.New("rocketmq not support")
+ }
+
+ return c, nil
+}
+
+type Client struct {
+ ctx context.Context
+ producerFacade ProducerFacade
+ kafkaConsumerConfig KafkaConsumerConfig
+}
+
+func (c Client) Apply() error {
+ panic("implement me")
+}
+
+func (c Client) Close() error {
+ return nil
+}
+
+func (c Client) Call(req *client.Request) (res interface{}, err error) {
+ body, err := ioutil.ReadAll(req.IngressRequest.Body)
+ if err != nil {
+ return nil, err
+ }
+
+ paths := strings.Split(req.API.Path, "/")
+ if len(paths) < 3 {
+ return nil, perrors.New("failed to send message, broker or Topic not found")
+ }
+
+ switch MQActionStrToInt[paths[0]] {
+ case MQActionPublish:
+ var pReq MQProduceRequest
+ err = json.Unmarshal(body, &pReq)
+ if err != nil {
+ return nil, err
+ }
+ err = c.producerFacade.Send(pReq.Msg, WithTopic(pReq.Topic))
+ if err != nil {
+ return nil, err
+ }
+ case MQActionSubscribe:
+ var cReq MQSubscribeRequest
+ err = json.Unmarshal(body, &cReq)
+ if err != nil {
+ return nil, err
+ }
+ if _, ok := consumerFacadeMap.Load(cReq.ConsumerGroup); !ok {
Review comment:
有数据说明已经订阅 考虑幂等性不要care了
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go-pixiu] ztelur commented on a change in pull request #300: [feature eventmesh]eventmesh
Posted by GitBox <gi...@apache.org>.
ztelur commented on a change in pull request #300:
URL: https://github.com/apache/dubbo-go-pixiu/pull/300#discussion_r752171952
##########
File path: pkg/client/mq/mq.go
##########
@@ -0,0 +1,156 @@
+/*
+ * 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 mq
+
+import (
+ "context"
+ "encoding/json"
+ "io/ioutil"
+ "strings"
+ "sync"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/client"
+ "github.com/apache/dubbo-go-pixiu/pkg/common/constant"
+ "github.com/apache/dubbo-go-pixiu/pkg/logger"
+)
+
+import (
+ perrors "github.com/pkg/errors"
+)
+
+var (
+ mqClient *Client
+ once sync.Once
+ consumerFacadeMap sync.Map
+)
+
+func NewSingletonMQClient(config Config) *Client {
+ if mqClient == nil {
+ once.Do(func() {
+ var err error
+ mqClient, err = NewMQClient(config)
+ if err != nil {
+ logger.Errorf("create mq client failed, %s", err.Error())
+ }
+ })
+ }
+ return mqClient
+}
+
+func NewMQClient(config Config) (*Client, error) {
+ var c *Client
+ ctx := context.Background()
+ switch config.MqType {
+ case constant.MQTypeKafka:
+ pf, err := NewKafkaProviderFacade(config.KafkaProducerConfig)
+ if err != nil {
+ return nil, err
+ }
+ c = &Client{
+ ctx: ctx,
+ producerFacade: pf,
+ kafkaConsumerConfig: config.KafkaConsumerConfig,
+ }
+ case constant.MQTypeRocketMQ:
+ return nil, perrors.New("rocketmq not support")
+ }
+
+ return c, nil
+}
+
+type Client struct {
+ ctx context.Context
+ producerFacade ProducerFacade
+ kafkaConsumerConfig KafkaConsumerConfig
+}
+
+func (c Client) Apply() error {
+ panic("implement me")
+}
+
+func (c Client) Close() error {
+ return nil
+}
+
+func (c Client) Call(req *client.Request) (res interface{}, err error) {
+ body, err := ioutil.ReadAll(req.IngressRequest.Body)
+ if err != nil {
+ return nil, err
+ }
+
+ paths := strings.Split(req.API.Path, "/")
+ if len(paths) < 3 {
+ return nil, perrors.New("failed to send message, broker or Topic not found")
+ }
+
+ switch MQActionStrToInt[paths[0]] {
+ case MQActionPublish:
+ var pReq MQProduceRequest
+ err = json.Unmarshal(body, &pReq)
+ if err != nil {
+ return nil, err
+ }
+ err = c.producerFacade.Send(pReq.Msg, WithTopic(pReq.Topic))
+ if err != nil {
+ return nil, err
+ }
+ case MQActionSubscribe:
+ var cReq MQSubscribeRequest
+ err = json.Unmarshal(body, &cReq)
+ if err != nil {
+ return nil, err
+ }
+ if _, ok := consumerFacadeMap.Load(cReq.ConsumerGroup); !ok {
Review comment:
Load有数据的分支是不是遗漏了
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go-pixiu] tydhot commented on a change in pull request #300: [feature eventmesh]eventmesh
Posted by GitBox <gi...@apache.org>.
tydhot commented on a change in pull request #300:
URL: https://github.com/apache/dubbo-go-pixiu/pull/300#discussion_r753807550
##########
File path: pkg/client/mq/kafka_facade.go
##########
@@ -0,0 +1,256 @@
+/*
+ * 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 mq
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "fmt"
+ "net/http"
+ "strconv"
+ "sync"
+ "time"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/logger"
+)
+
+import (
+ "github.com/Shopify/sarama"
+ perrors "github.com/pkg/errors"
+)
+
+type kafkaErrors struct {
+ count int
+ err string
+}
+
+func (ke kafkaErrors) Error() string {
+ return fmt.Sprintf("Failed to deliver %d messages due to %s", ke.count, ke.err)
+}
+
+func NewKafkaConsumerFacade(config KafkaConsumerConfig, consumerGroup string) (*KafkaConsumerFacade, error) {
+ c := sarama.NewConfig()
+ c.ClientID = config.ClientID
+ c.Metadata.Full = config.Metadata.Full
+ c.Metadata.Retry.Max = config.Metadata.Retry.Max
+ c.Metadata.Retry.Backoff = config.Metadata.Retry.Backoff
+ if config.ProtocolVersion != "" {
+ version, err := sarama.ParseKafkaVersion(config.ProtocolVersion)
+ if err != nil {
+ return nil, err
+ }
+ c.Version = version
+ }
+ client, err := sarama.NewConsumerGroup(config.Brokers, consumerGroup, c)
+ if err != nil {
+ return nil, err
+ }
+
+ return &KafkaConsumerFacade{consumerGroup: client, httpClient: &http.Client{Timeout: 5 * time.Second}, done: make(chan struct{})}, nil
+}
+
+type KafkaConsumerFacade struct {
+ consumerGroup sarama.ConsumerGroup
+ consumerManager map[string]func()
+ httpClient *http.Client
+ wg sync.WaitGroup
+ done chan struct{}
+}
+
+func (f *KafkaConsumerFacade) Subscribe(ctx context.Context, opts ...Option) error {
+ cOpt := DefaultOptions()
+ cOpt.ApplyOpts(opts...)
+ c, cancel := context.WithCancel(ctx)
+ key := GetConsumerManagerKey(cOpt.TopicList, cOpt.ConsumerGroup)
Review comment:
这里我和俊铭的考虑是 这一块其实是在kafka之上的一个通用的facade抽象。目前因为只有一个kafka实现可能比较冗余，但是在多种实现下，这里可能需要把一些通用参数转化成每种mq能够进行必要性判断和具体使用的格式，这里主要是出于这样 的考虑不过MQOptions也应该移动到kafka_facade内
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go-pixiu] tydhot commented on a change in pull request #300: [feature eventmesh]eventmesh
Posted by GitBox <gi...@apache.org>.
tydhot commented on a change in pull request #300:
URL: https://github.com/apache/dubbo-go-pixiu/pull/300#discussion_r753801331
##########
File path: pkg/client/mq/facade.go
##########
@@ -0,0 +1,92 @@
+/*
+ * 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 mq
+
+import (
+ "context"
+ "strings"
+)
+
+type ConsumerFacade interface {
+ // Subscribe message with specified broker and Topic, then handle msg with handler which send msg to real consumers
+ Subscribe(ctx context.Context, option ...Option) error
+ UnSubscribe(opts ...Option) error
+ Stop()
+}
+
+func GetConsumerManagerKey(topic []string, consumerGroup string) string {
+ return strings.Join((topic), ".") + "_" + consumerGroup
+}
+
+// MQOptions Consumer options
+// TODO: Add rocketmq params
+type MQOptions struct {
+ TopicList []string
+ ConsumeUrl string
+ CheckUrl string
+ ConsumerGroup string
+}
+
+func (o *MQOptions) ApplyOpts(opts ...Option) {
+ for _, opt := range opts {
+ opt(o)
+ }
+}
+
+func DefaultOptions() *MQOptions {
+ return &MQOptions{
+ TopicList: []string{"demo-topic"},
Review comment:
嗯 这个缺省项意义不大
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go-pixiu] codecov-commenter edited a comment on pull request #300: [feature eventmesh]eventmesh
Posted by GitBox <gi...@apache.org>.
codecov-commenter edited a comment on pull request #300:
URL: https://github.com/apache/dubbo-go-pixiu/pull/300#issuecomment-968298380
# [Codecov](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#300](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (045ee71) into [develop](https://codecov.io/gh/apache/dubbo-go-pixiu/commit/bf6dde05c14d394a5f6bda919fc6473107ae8d25?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (bf6dde0) will **not change** coverage.
> The diff coverage is `n/a`.
> :exclamation: Current head 045ee71 differs from pull request most recent head aa2ada5. Consider uploading reports for the commit aa2ada5 to get more accurate results
[](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## develop #300 +/- ##
========================================
Coverage 36.04% 36.04%
========================================
Files 55 55
Lines 3299 3299
========================================
Hits 1189 1189
Misses 1991 1991
Partials 119 119
```
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [bf6dde0...aa2ada5](https://codecov.io/gh/apache/dubbo-go-pixiu/pull/300?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go-pixiu] AlexStocks commented on a change in pull request #300: [feature eventmesh]eventmesh
Posted by GitBox <gi...@apache.org>.
AlexStocks commented on a change in pull request #300:
URL: https://github.com/apache/dubbo-go-pixiu/pull/300#discussion_r749007512
##########
File path: pkg/client/mq/kafka_facade.go
##########
@@ -0,0 +1,237 @@
+/*
+ * 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 mq
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "net/http"
+ "strconv"
+ "sync"
+ "time"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/filter/event"
+ "github.com/apache/dubbo-go-pixiu/pkg/logger"
+)
+
+import (
+ "github.com/Shopify/sarama"
+ perrors "github.com/pkg/errors"
+)
+
+func NewKafkaConsumerFacade(config event.KafkaConsumerConfig) (*KafkaConsumerFacade, error) {
+ c := sarama.NewConfig()
+ c.ClientID = config.ClientID
+ c.Metadata.Full = config.Metadata.Full
+ c.Metadata.Retry.Max = config.Metadata.Retry.Max
+ c.Metadata.Retry.Backoff = config.Metadata.Retry.Backoff
+ if config.ProtocolVersion != "" {
+ version, err := sarama.ParseKafkaVersion(config.ProtocolVersion)
+ if err != nil {
+ return nil, err
+ }
+ c.Version = version
+ }
+ consumer, err := sarama.NewConsumer(config.Brokers, c)
+ if err != nil {
+ return nil, err
+ }
+
+ return &KafkaConsumerFacade{consumer: consumer, httpClient: &http.Client{Timeout: 5 * time.Second}, done: make(chan struct{})}, nil
+}
+
+type KafkaConsumerFacade struct {
+ consumer sarama.Consumer
+ consumerManager map[string]func()
+ rwLock sync.RWMutex
+ httpClient *http.Client
+ wg sync.WaitGroup
+ done chan struct{}
+}
+
+func (f *KafkaConsumerFacade) Subscribe(ctx context.Context, opts ...Option) error {
+ cOpt := DefaultOptions()
+ cOpt.ApplyOpts(opts...)
+ partConsumer, err := f.consumer.ConsumePartition(cOpt.Topic, int32(cOpt.Partition), sarama.OffsetOldest)
+ if err != nil {
+ return err
+ }
+ c, cancel := context.WithCancel(ctx)
+ key := GetConsumerManagerKey(cOpt.Topic, int32(cOpt.Partition))
+ f.rwLock.Lock()
+ defer f.rwLock.Unlock()
+ f.consumerManager[key] = cancel
+ f.wg.Add(2)
+ go f.ConsumePartitions(c, partConsumer, cOpt.ConsumeUrl)
+ go f.checkConsumerIsAlive(c, key, cOpt.CheckUrl)
+ return nil
+}
+
+// ConsumePartitions consume function
+func (f *KafkaConsumerFacade) ConsumePartitions(ctx context.Context, partConsumer sarama.PartitionConsumer, consumeUrl string) {
+ defer f.wg.Done()
+
+ for {
+ select {
+ case <-f.done:
+ logger.Info()
+ case msg := <-partConsumer.Messages():
+ data, err := json.Marshal(event.MQMsgPush{Msg: []string{string(msg.Value)}})
+ if err != nil {
+ logger.Warn()
+ continue
+ }
+
+ req, err := http.NewRequest(http.MethodPost, consumeUrl, bytes.NewReader(data))
+ if err != nil {
+ logger.Warn()
+ continue
+ }
+
+ for i := 0; i < 5; i++ {
+ err := func() error {
+ resp, err := f.httpClient.Do(req)
+ if err != nil {
+ return err
+ }
+ defer resp.Body.Close()
+
+ if resp.StatusCode == http.StatusOK {
+ return nil
+ }
+ return perrors.New("failed send msg to consumer with status code " + strconv.Itoa(resp.StatusCode))
Review comment:
failed send msg to consumer
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go-pixiu] tydhot commented on a change in pull request #300: [feature eventmesh]eventmesh
Posted by GitBox <gi...@apache.org>.
tydhot commented on a change in pull request #300:
URL: https://github.com/apache/dubbo-go-pixiu/pull/300#discussion_r749036145
##########
File path: pkg/client/mq/kafka_facade.go
##########
@@ -0,0 +1,237 @@
+/*
+ * 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 mq
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "net/http"
+ "strconv"
+ "sync"
+ "time"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/filter/event"
+ "github.com/apache/dubbo-go-pixiu/pkg/logger"
+)
+
+import (
+ "github.com/Shopify/sarama"
+ perrors "github.com/pkg/errors"
+)
+
+func NewKafkaConsumerFacade(config event.KafkaConsumerConfig) (*KafkaConsumerFacade, error) {
+ c := sarama.NewConfig()
+ c.ClientID = config.ClientID
+ c.Metadata.Full = config.Metadata.Full
+ c.Metadata.Retry.Max = config.Metadata.Retry.Max
+ c.Metadata.Retry.Backoff = config.Metadata.Retry.Backoff
+ if config.ProtocolVersion != "" {
+ version, err := sarama.ParseKafkaVersion(config.ProtocolVersion)
+ if err != nil {
+ return nil, err
+ }
+ c.Version = version
+ }
+ consumer, err := sarama.NewConsumer(config.Brokers, c)
+ if err != nil {
+ return nil, err
+ }
+
+ return &KafkaConsumerFacade{consumer: consumer, httpClient: &http.Client{Timeout: 5 * time.Second}, done: make(chan struct{})}, nil
+}
+
+type KafkaConsumerFacade struct {
+ consumer sarama.Consumer
+ consumerManager map[string]func()
+ rwLock sync.RWMutex
+ httpClient *http.Client
+ wg sync.WaitGroup
+ done chan struct{}
+}
+
+func (f *KafkaConsumerFacade) Subscribe(ctx context.Context, opts ...Option) error {
+ cOpt := DefaultOptions()
+ cOpt.ApplyOpts(opts...)
+ partConsumer, err := f.consumer.ConsumePartition(cOpt.Topic, int32(cOpt.Partition), sarama.OffsetOldest)
+ if err != nil {
+ return err
+ }
+ c, cancel := context.WithCancel(ctx)
+ key := GetConsumerManagerKey(cOpt.Topic, int32(cOpt.Partition))
+ f.rwLock.Lock()
+ defer f.rwLock.Unlock()
+ f.consumerManager[key] = cancel
+ f.wg.Add(2)
+ go f.ConsumePartitions(c, partConsumer, cOpt.ConsumeUrl)
+ go f.checkConsumerIsAlive(c, key, cOpt.CheckUrl)
+ return nil
+}
+
+// ConsumePartitions consume function
+func (f *KafkaConsumerFacade) ConsumePartitions(ctx context.Context, partConsumer sarama.PartitionConsumer, consumeUrl string) {
+ defer f.wg.Done()
+
+ for {
+ select {
+ case <-f.done:
+ logger.Info()
+ case msg := <-partConsumer.Messages():
+ data, err := json.Marshal(event.MQMsgPush{Msg: []string{string(msg.Value)}})
+ if err != nil {
+ logger.Warn()
+ continue
+ }
+
+ req, err := http.NewRequest(http.MethodPost, consumeUrl, bytes.NewReader(data))
+ if err != nil {
+ logger.Warn()
+ continue
+ }
+
+ for i := 0; i < 5; i++ {
+ err := func() error {
+ resp, err := f.httpClient.Do(req)
+ if err != nil {
+ return err
+ }
+ defer resp.Body.Close()
+
+ if resp.StatusCode == http.StatusOK {
+ return nil
+ }
+ return perrors.New("failed send msg to consumer with status code " + strconv.Itoa(resp.StatusCode))
Review comment:
got it
##########
File path: pkg/client/mq/kafka_facade.go
##########
@@ -0,0 +1,237 @@
+/*
+ * 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 mq
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "net/http"
+ "strconv"
+ "sync"
+ "time"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/filter/event"
+ "github.com/apache/dubbo-go-pixiu/pkg/logger"
+)
+
+import (
+ "github.com/Shopify/sarama"
+ perrors "github.com/pkg/errors"
+)
+
+func NewKafkaConsumerFacade(config event.KafkaConsumerConfig) (*KafkaConsumerFacade, error) {
+ c := sarama.NewConfig()
+ c.ClientID = config.ClientID
+ c.Metadata.Full = config.Metadata.Full
+ c.Metadata.Retry.Max = config.Metadata.Retry.Max
+ c.Metadata.Retry.Backoff = config.Metadata.Retry.Backoff
+ if config.ProtocolVersion != "" {
+ version, err := sarama.ParseKafkaVersion(config.ProtocolVersion)
+ if err != nil {
+ return nil, err
+ }
+ c.Version = version
+ }
+ consumer, err := sarama.NewConsumer(config.Brokers, c)
+ if err != nil {
+ return nil, err
+ }
+
+ return &KafkaConsumerFacade{consumer: consumer, httpClient: &http.Client{Timeout: 5 * time.Second}, done: make(chan struct{})}, nil
+}
+
+type KafkaConsumerFacade struct {
+ consumer sarama.Consumer
+ consumerManager map[string]func()
+ rwLock sync.RWMutex
+ httpClient *http.Client
+ wg sync.WaitGroup
+ done chan struct{}
+}
+
+func (f *KafkaConsumerFacade) Subscribe(ctx context.Context, opts ...Option) error {
+ cOpt := DefaultOptions()
+ cOpt.ApplyOpts(opts...)
+ partConsumer, err := f.consumer.ConsumePartition(cOpt.Topic, int32(cOpt.Partition), sarama.OffsetOldest)
+ if err != nil {
+ return err
+ }
+ c, cancel := context.WithCancel(ctx)
+ key := GetConsumerManagerKey(cOpt.Topic, int32(cOpt.Partition))
+ f.rwLock.Lock()
+ defer f.rwLock.Unlock()
+ f.consumerManager[key] = cancel
+ f.wg.Add(2)
+ go f.ConsumePartitions(c, partConsumer, cOpt.ConsumeUrl)
+ go f.checkConsumerIsAlive(c, key, cOpt.CheckUrl)
+ return nil
+}
+
+// ConsumePartitions consume function
+func (f *KafkaConsumerFacade) ConsumePartitions(ctx context.Context, partConsumer sarama.PartitionConsumer, consumeUrl string) {
+ defer f.wg.Done()
+
+ for {
+ select {
+ case <-f.done:
+ logger.Info()
Review comment:
ok
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org