add relay

mqtt/client.go

@@ -0,0 +1,73 @@
+package mqtt
+
+import (
+	"crypto/tls"
+	"fmt"
+	"log"
+
+	mqtt "github.com/eclipse/paho.mqtt.golang"
+	mqttClient "github.com/eclipse/paho.mqtt.golang"
+	"github.com/ssp97/mqtt_relay/config"
+)
+
+var globalCfg *config.Config
+
+func MqttClientInit(cfg *config.Config) {
+	globalCfg = cfg
+}
+
+func MqttClientConnect(session *Session) (err error) {
+	tlsConfig := &tls.Config{
+		InsecureSkipVerify: true,
+		ClientAuth:         tls.NoClientCert,
+	}
+	opts := mqttClient.NewClientOptions()
+	opts.AddBroker(globalCfg.Relay.Broker)
+	opts.SetClientID(session.ClientId)
+	opts.SetTLSConfig(tlsConfig)
+
+	opts.SetUsername(string(session.Username))
+	opts.SetPassword(string(session.Password))
+
+	opts.OnConnect = func(c mqttClient.Client) {
+		fmt.Println("Connected to MQTT broker")
+	}
+
+	client := mqttClient.NewClient(opts)
+	session.Client = &client
+	if token := client.Connect(); token.Wait() && token.Error() != nil {
+		log.Printf("Failed to connect to MQTT broker: %v\n", token.Error())
+		err = token.Error()
+	}
+
+	return
+}
+
+func MqttClientDisconnect(session *Session) {
+	client := *session.Client
+	client.Disconnect(0)
+}
+
+func MqttClientSubscribe(session *Session, topic string, qos byte) {
+	client := *session.Client
+	client.Subscribe(topic, qos, func(client mqtt.Client, msg mqtt.Message) {
+		log.Printf("Received message on topic %s: %s\n", msg.Topic(), msg.Payload())
+		server := GetMqttServer()
+		err := server.Publish(msg.Topic(), msg.Payload(), msg.Retained(), msg.Qos())
+		if err != nil {
+			log.Println("server.Publish err=", err)
+		}
+
+	})
+
+}
+
+func MqttClientUnsubscribe(session *Session, topic string) {
+	client := *session.Client
+	client.Unsubscribe(topic)
+}
+
+func MqttClientPublish(session *Session, topic string, qos byte, retained bool, payload interface{}) {
+	client := *session.Client
+	client.Publish(topic, qos, retained, payload)
+}

mqtt/hook.go

@@ -35,7 +35,7 @@ func (h *MqttServerHook) Provides(b byte) bool {
 }
 
 func (h *MqttServerHook) Init(config any) error {
-	h.Log.Info("initialised")
+	h.Log.Info("hook was initialised")
 	if _, ok := config.(*MqttServerHookOptions); !ok && config != nil {
 		return mqttServerV2.ErrInvalidConfigType
 	}
@@ -55,14 +55,18 @@ func (h *MqttServerHook) subscribeCallback(cl *mqttServerV2.Client, sub packets.
 func (h *MqttServerHook) OnConnect(cl *mqttServerV2.Client, pk packets.Packet) error {
 	h.Log.Info("client connected", "client", cl.ID)
 
+	// session := GetSession(cl.ID)
+	// if session != nil {
+
+	// }
 	// Example demonstrating how to subscribe to a topic within the hook.
-	h.config.Server.Subscribe("hook/direct/publish", 1, h.subscribeCallback)
+	//h.config.Server.Subscribe("hook/direct/publish", 1, h.subscribeCallback)
 
 	// Example demonstrating how to publish a message within the hook
-	err := h.config.Server.Publish("hook/direct/publish", []byte("packet hook message"), false, 0)
-	if err != nil {
-		h.Log.Error("hook.publish", "error", err)
-	}
+	//err := h.config.Server.Publish("hook/direct/publish", []byte("packet hook message"), false, 0)
+	//if err != nil {
+	//	h.Log.Error("hook.publish", "error", err)
+	//}
 
 	return nil
 }
@@ -78,22 +82,29 @@ func (h *MqttServerHook) OnDisconnect(cl *mqttServerV2.Client, err error, expire
 
 func (h *MqttServerHook) OnSubscribed(cl *mqttServerV2.Client, pk packets.Packet, reasonCodes []byte) {
 	h.Log.Info(fmt.Sprintf("subscribed qos=%v", reasonCodes), "client", cl.ID, "filters", pk.Filters)
+	session := GetSession(cl.ID)
+	for _, v := range pk.Filters {
+		MqttClientSubscribe(session, v.Filter, reasonCodes[0])
+	}
+
 }
 
 func (h *MqttServerHook) OnUnsubscribed(cl *mqttServerV2.Client, pk packets.Packet) {
 	h.Log.Info("unsubscribed", "client", cl.ID, "filters", pk.Filters)
+	session := GetSession(cl.ID)
+	for _, v := range pk.Filters {
+		MqttClientUnsubscribe(session, v.Filter)
+	}
 }
 
 func (h *MqttServerHook) OnPublish(cl *mqttServerV2.Client, pk packets.Packet) (packets.Packet, error) {
-	h.Log.Info("received from client", "client", cl.ID, "payload", string(pk.Payload))
-
-	pkx := pk
-	if string(pk.Payload) == "hello" {
-		pkx.Payload = []byte("hello world")
-		h.Log.Info("received modified packet from client", "client", cl.ID, "payload", string(pkx.Payload))
+	session := GetSession(cl.ID)
+	if session != nil {
+		h.Log.Info("received from client", "client", cl.ID, "payload", string(pk.Payload))
+		MqttClientPublish(session, pk.TopicName, pk.FixedHeader.Qos, pk.FixedHeader.Retain, pk.Payload)
 	}
 
-	return pkx, nil
+	return pk, nil
 }
 
 func (h *MqttServerHook) OnPublished(cl *mqttServerV2.Client, pk packets.Packet) {
@@ -103,6 +114,11 @@ func (h *MqttServerHook) OnPublished(cl *mqttServerV2.Client, pk packets.Packet)
 // OnConnectAuthenticate returns true/allowed for all requests.
 func (h *MqttServerHook) OnConnectAuthenticate(cl *mqttServerV2.Client, pk packets.Packet) bool {
 	h.Log.Info("Auth", pk.Connect.Username, pk.Connect.Password)
+	session := CreateSession(cl.ID, pk.Connect.Username, pk.Connect.Password)
+	err := MqttClientConnect(session)
+	if err != nil {
+		return false
+	}
 	return true
 }
 

mqtt/server.go

@@ -11,8 +11,16 @@ import (
 	"github.com/ssp97/mqtt_relay/config"
 )
 
+var globalServer *mqttServerV2.Server
+
+func GetMqttServer() *mqttServerV2.Server {
+	return globalServer
+}
+
 func MqttServerStart(cfg *config.Config) *mqttServerV2.Server {
-	server := mqttServerV2.New(nil)
+	server := mqttServerV2.New(&mqttServerV2.Options{
+		InlineClient: true,
+	})
 	// _ = server.AddHook(new(auth.AllowHook), nil)
 
 	var tlsConfig *tls.Config
@@ -52,5 +60,6 @@ func MqttServerStart(cfg *config.Config) *mqttServerV2.Server {
 		}
 	}()
 
+	globalServer = server
 	return server
 }

mqtt/session.go

@@ -1,6 +1,25 @@
 package mqtt
 
+import mqttClient "github.com/eclipse/paho.mqtt.golang"
+
 type Session struct {
-	Username string
-	Password string
+	ClientId string
+	Username []byte
+	Password []byte
+	Client   *mqttClient.Client
+}
+
+var Sessions map[string]*Session = make(map[string]*Session)
+
+func CreateSession(mqttId string, user, passwd []byte) *Session {
+	Sessions[mqttId] = &Session{
+		ClientId: mqttId,
+		Username: user,
+		Password: passwd,
+	}
+	return Sessions[mqttId]
+}
+
+func GetSession(mqttId string) *Session {
+	return Sessions[mqttId]
 }