Add KVMap for MQTTContext and update Kafka filter (#462)

* add kvmap to mqttcontext, allow kafka filter get data from mqtt context kvmap

* update kafka spec to eliminate ambiguity

* remove meanless constant

pkg/context/mqttcontext.go

@@ -41,15 +41,17 @@ type (
  DisconnectPacket() *packets.DisconnectPacket // read only
  SubscribePacket() *packets.SubscribePacket // read only
  UnsubscribePacket() *packets.UnsubscribePacket // read only
-
- PublishPacket() MQTTPublishPacket
+ PublishPacket() *packets.PublishPacket // read only
 
  SetDrop() // set drop value to true
  Drop() bool // if true, this mqtt packet will be dropped
  SetDisconnect() // set disconnect value to true
  Disconnect() bool // if true, this mqtt client will be disconnected
  SetEarlyStop() // set early stop value to true
  EarlyStop() bool // if early stop is true, pipeline will skip following filters and return
+
+ SetKV(string, interface{})
+ GetKV(string) interface{}
  }
 
  // MQTTClient contains client info that send this packet
@@ -74,8 +76,7 @@ type (
  client MQTTClient
  packet packets.ControlPacket
  packetType MQTTPacketType
-
- publishPacket MQTTPublishPacket
+ kvMap map[string]interface{}
 
  err error
  drop int32
@@ -85,7 +86,7 @@ type (
 
  // MQTTResult is result for handling mqtt request
  MQTTResult struct {
- Err error
+ ErrString string
  }
 )
 
@@ -120,14 +121,14 @@ func NewMQTTContext(ctx stdcontext.Context, client MQTTClient, packet packets.Co
  cancelFunc: cancelFunc,
  startTime: startTime,
  client: client,
+ kvMap: make(map[string]interface{}),
  }
 
- switch packet := packet.(type) {
+ switch packet.(type) {
  case *packets.ConnectPacket:
  mqttCtx.packetType = MQTTConnect
  case *packets.PublishPacket:
  mqttCtx.packetType = MQTTPublish
- mqttCtx.publishPacket = newMQTTPublishPacket(packet)
  case *packets.DisconnectPacket:
  mqttCtx.packetType = MQTTDisconnect
  case *packets.SubscribePacket:
@@ -223,8 +224,8 @@ func (ctx *mqttContext) ConnectPacket() *packets.ConnectPacket {
  return ctx.packet.(*packets.ConnectPacket)
 }
 
-func (ctx *mqttContext) PublishPacket() MQTTPublishPacket {
- return ctx.publishPacket
+func (ctx *mqttContext) PublishPacket() *packets.PublishPacket {
+ return ctx.packet.(*packets.PublishPacket)
 }
 
 func (ctx *mqttContext) DisconnectPacket() *packets.DisconnectPacket {
@@ -262,3 +263,11 @@ func (ctx *mqttContext) SetEarlyStop() {
 func (ctx *mqttContext) EarlyStop() bool {
  return atomic.LoadInt32(&ctx.earlyStop) == 1
 }
+
+func (ctx *mqttContext) SetKV(key string, value interface{}) {
+ ctx.kvMap[key] = value
+}
+
+func (ctx *mqttContext) GetKV(key string) interface{} {
+ return ctx.kvMap[key]
+}

pkg/filter/connectcontrol/connectcontrol.go

@@ -18,7 +18,6 @@
 package connectcontrol
 
 import (
- "errors"
  "regexp"
 
  "github.com/megaease/easegress/pkg/context"
@@ -34,7 +33,6 @@ const (
 )
 
 // ErrBannedClientOrTopic is error for banned client or topic
-var ErrBannedClientOrTopic = errors.New(resultBannedClientOrTopic)
 
 func init() {
  pipeline.Register(&ConnectControl{})
@@ -162,7 +160,7 @@ func (cc *ConnectControl) checkBan(ctx context.MQTTContext) bool {
  if _, ok := cc.bannedClients[cid]; ok {
  return true
  }
- topic := ctx.PublishPacket().Topic()
+ topic := ctx.PublishPacket().TopicName
  if cc.bannedTopicRe != nil && cc.bannedTopicRe.MatchString(topic) {
  return true
  }
@@ -183,7 +181,7 @@ func (cc *ConnectControl) HandleMQTT(ctx context.MQTTContext) *context.MQTTResul
  if cc.spec.EarlyStop {
  ctx.SetEarlyStop()
  }
- return &context.MQTTResult{Err: ErrBannedClientOrTopic}
+ return &context.MQTTResult{ErrString: resultBannedClientOrTopic}
  }
  return &context.MQTTResult{}
 }

pkg/filter/connectcontrol/connectcontrol_test.go

@@ -96,7 +96,7 @@ func TestConnectControl(t *testing.T) {
 type testCase struct {
  cid string
  topic string
- err error
+ errString string
  disconnect bool
  earlyStop bool
 }
@@ -110,7 +110,7 @@ func doTest(t *testing.T, spec *Spec, testCases []testCase) {
  for _, test := range testCases {
  ctx := newContext(test.cid, test.topic)
  res := cc.HandleMQTT(ctx)
- assert.Equal(res.Err, test.err)
+ assert.Equal(res.ErrString, test.errString)
  assert.Equal(ctx.Disconnect(), test.disconnect)
  assert.Equal(ctx.EarlyStop(), test.earlyStop)
  }
@@ -128,11 +128,11 @@ func TestHandleMQTT(t *testing.T) {
  EarlyStop: true,
  }
  testCases := []testCase{
- {cid: "ban1", topic: "unban", err: errors.New(resultBannedClientOrTopic), disconnect: true, earlyStop: true},
- {cid: "ban2", topic: "unban", err: errors.New(resultBannedClientOrTopic), disconnect: true, earlyStop: true},
- {cid: "unban1", topic: "ban/sport/ball", err: nil, disconnect: false, earlyStop: false},
- {cid: "unban2", topic: "ban/sport/run", err: nil, disconnect: false, earlyStop: false},
- {cid: "unban", topic: "unban", err: nil, disconnect: false, earlyStop: false},
+ {cid: "ban1", topic: "unban", errString: resultBannedClientOrTopic, disconnect: true, earlyStop: true},
+ {cid: "ban2", topic: "unban", errString: resultBannedClientOrTopic, disconnect: true, earlyStop: true},
+ {cid: "unban1", topic: "ban/sport/ball", errString: "", disconnect: false, earlyStop: false},
+ {cid: "unban2", topic: "ban/sport/run", errString: "", disconnect: false, earlyStop: false},
+ {cid: "unban", topic: "unban", errString: "", disconnect: false, earlyStop: false},
  }
  doTest(t, spec, testCases)
 
@@ -142,10 +142,10 @@ func TestHandleMQTT(t *testing.T) {
  EarlyStop: true,
  }
  testCases = []testCase{
- {cid: "unban1", topic: "ban/sport/ball", err: errors.New(resultBannedClientOrTopic), disconnect: true, earlyStop: true},
- {cid: "unban2", topic: "ban/sport/run", err: errors.New(resultBannedClientOrTopic), disconnect: true, earlyStop: true},
- {cid: "unban3", topic: "unban/sport", err: nil, disconnect: false, earlyStop: false},
- {cid: "unban4", topic: "unban", err: nil, disconnect: false, earlyStop: false},
+ {cid: "unban1", topic: "ban/sport/ball", errString: resultBannedClientOrTopic, disconnect: true, earlyStop: true},
+ {cid: "unban2", topic: "ban/sport/run", errString: resultBannedClientOrTopic, disconnect: true, earlyStop: true},
+ {cid: "unban3", topic: "unban/sport", errString: "", disconnect: false, earlyStop: false},
+ {cid: "unban4", topic: "unban", errString: "", disconnect: false, earlyStop: false},
  }
  doTest(t, spec, testCases)
 
@@ -155,10 +155,10 @@ func TestHandleMQTT(t *testing.T) {
  EarlyStop: true,
  }
  testCases = []testCase{
- {cid: "phone123", topic: "ban/sport/ball", err: errors.New(resultBannedClientOrTopic), disconnect: true, earlyStop: true},
- {cid: "phone256", topic: "ban/sport/run", err: errors.New(resultBannedClientOrTopic), disconnect: true, earlyStop: true},
- {cid: "tv", topic: "unban/sport", err: nil, disconnect: false, earlyStop: false},
- {cid: "tv", topic: "unban", err: nil, disconnect: false, earlyStop: false},
+ {cid: "phone123", topic: "ban/sport/ball", errString: resultBannedClientOrTopic, disconnect: true, earlyStop: true},
+ {cid: "phone256", topic: "ban/sport/run", errString: resultBannedClientOrTopic, disconnect: true, earlyStop: true},
+ {cid: "tv", topic: "unban/sport", errString: "", disconnect: false, earlyStop: false},
+ {cid: "tv", topic: "unban", errString: "", disconnect: false, earlyStop: false},
  }
  doTest(t, spec, testCases)
 
@@ -168,10 +168,10 @@ func TestHandleMQTT(t *testing.T) {
  EarlyStop: true,
  }
  testCases = []testCase{
- {cid: "phone123", topic: "ban/sport/ball", err: errors.New(resultBannedClientOrTopic), disconnect: true, earlyStop: true},
- {cid: "phone256", topic: "ban/sport/run", err: errors.New(resultBannedClientOrTopic), disconnect: true, earlyStop: true},
- {cid: "tv", topic: "unban", err: nil, disconnect: false, earlyStop: false},
- {cid: "tv", topic: "unban", err: nil, disconnect: false, earlyStop: false},
+ {cid: "phone123", topic: "ban/sport/ball", errString: resultBannedClientOrTopic, disconnect: true, earlyStop: true},
+ {cid: "phone256", topic: "ban/sport/run", errString: resultBannedClientOrTopic, disconnect: true, earlyStop: true},
+ {cid: "tv", topic: "unban", errString: "", disconnect: false, earlyStop: false},
+ {cid: "tv", topic: "unban", errString: "", disconnect: false, earlyStop: false},
  }
  doTest(t, spec, testCases)