update rawtrafficcontroller kind switch to support mqttproxy (easegre…

…ss-io#694)

* update rawtrafficcontroller kind switch to support mqttproxy

* update rawtrafficcontroller kind switch to support mqttproxy

* integration test

* update integration test

build/test/integration_test.go

@@ -24,7 +24,9 @@ import (
  "net/http"
  "strings"
  "testing"
+ "time"
 
+ paho "github.com/eclipse/paho.mqtt.golang"
  "github.com/stretchr/testify/assert"
  "github.com/stretchr/testify/require"
 )
@@ -237,3 +239,35 @@ filters:
  assert.Nil(err)
  assert.Equal("hello from backend", string(data))
 }
+
+func getMQTTClient(clientID, userName, password string) (paho.Client, error) {
+ opts := paho.NewClientOptions().AddBroker("tcp:https://0.0.0.0:1883").SetClientID(clientID).SetUsername(userName).SetPassword(password)
+ c := paho.NewClient(opts)
+ if token := c.Connect(); token.Wait() && token.Error() != nil {
+ return nil, token.Error()
+ }
+ return c, nil
+}
+
+func TestMQTTProxy(t *testing.T) {
+ assert := assert.New(t)
+
+ yamlStr := `
+kind: MQTTProxy
+name: mqttproxy-test
+port: 1883
+`
+ ok, msg := createObject(t, yamlStr)
+ assert.True(ok, msg)
+ defer deleteObject(t, "mqttproxy-test")
+
+ var err error
+ for i := 0; i < 10; i++ {
+ _, err = getMQTTClient("client1", "test", "test")
+ if err == nil {
+ break
+ }
+ time.Sleep(time.Second)
+ }
+ assert.Nil(err)
+}

pkg/object/rawconfigtrafficcontroller/rawconfigtrafficcontroller.go

@@ -22,7 +22,6 @@ import (
 
  "github.com/megaease/easegress/pkg/context"
  "github.com/megaease/easegress/pkg/logger"
- "github.com/megaease/easegress/pkg/object/httpserver"
  "github.com/megaease/easegress/pkg/object/pipeline"
  "github.com/megaease/easegress/pkg/object/trafficcontroller"
  "github.com/megaease/easegress/pkg/supervisor"
@@ -138,12 +137,11 @@ func (rctc *RawConfigTrafficController) handleEvent(event *supervisor.ObjectEnti
  var err error
 
  kind := entity.Spec().Kind()
- switch kind {
- case httpserver.Kind:
- err = rctc.tc.DeleteTrafficGate(DefaultNamespace, name)
- case pipeline.Kind:
+ if kind == pipeline.Kind {
  err = rctc.tc.DeletePipeline(DefaultNamespace, name)
- default:
+ } else if _, ok := supervisor.TrafficObjectKinds[kind]; ok {
+ err = rctc.tc.DeleteTrafficGate(DefaultNamespace, name)
+ } else {
  logger.Errorf("BUG: unexpected kind %T", kind)
  }
 
@@ -156,12 +154,11 @@ func (rctc *RawConfigTrafficController) handleEvent(event *supervisor.ObjectEnti
  var err error
 
  kind := entity.Spec().Kind()
- switch kind {
- case httpserver.Kind:
- _, err = rctc.tc.CreateTrafficGate(DefaultNamespace, entity)
- case pipeline.Kind:
+ if kind == pipeline.Kind {
  _, err = rctc.tc.CreatePipeline(DefaultNamespace, entity)
- default:
+ } else if _, ok := supervisor.TrafficObjectKinds[kind]; ok {
+ _, err = rctc.tc.CreateTrafficGate(DefaultNamespace, entity)
+ } else {
  logger.Errorf("BUG: unexpected kind %T", kind)
  }
 
@@ -174,12 +171,11 @@ func (rctc *RawConfigTrafficController) handleEvent(event *supervisor.ObjectEnti
  var err error
 
  kind := entity.Instance().Kind()
- switch kind {
- case httpserver.Kind:
- _, err = rctc.tc.UpdateTrafficGate(DefaultNamespace, entity)
- case pipeline.Kind:
+ if kind == pipeline.Kind {
  _, err = rctc.tc.UpdatePipeline(DefaultNamespace, entity)
- default:
+ } else if _, ok := supervisor.TrafficObjectKinds[kind]; ok {
+ _, err = rctc.tc.UpdateTrafficGate(DefaultNamespace, entity)
+ } else {
  logger.Errorf("BUG: unexpected kind %T", kind)
  }