Modify message before delivering to client

[drmagice]

Modify message before delivering to client

[maddie]

I have solved this problem with this workaround:

• Decompress the payload
• Generate a new packet.Message, return false for current message so it won't be forwarded
• Re-send the new packet.Message

I think there should be an easier way to achieve this.

[maddie]

Hmm.. this workaround doesn't work, the resend messages aren't being delivered somehow.

Is there a way to do this?

[maddie]

So it turns out resending the message within the OnMsgArrived hook doesn't work, somehow Publish() seems to be stuck. I put the message into a channel and handling it from another place where the server instance is accessible, from there Publish works just fine.

[DrmagicE]

The message in OnMsgArrived is readonly, you cannot modified it in OnMsgArrived. Publish should work in OnMsgArrived , can you show me your codes?

Modify message in OnMsgArrived seems reasonable. I will check if i can add this feature via OnMsgArrived.

[maddie]

I have a struct defined as follows:

type Plugin struct {
   service gmqtt.Server
}

And assigned service on Load(service gmqtt.Server) receiver function

func (p Plugin) Load(service gmqtt.Server) error {
    p.service = service
    return nil
}

Then in OnMsgArrived

func (p Plugin) OnMsgArrived(arrived gmqtt.OnMsgArrived) gmqtt.OnMsgArrived {
    return func(ctx context.Context, client gmqtt.Client, msg packets.Message) bool {
        if needsDecompression {
            // decompress msg.Payload()
            b := decompress(msg.Payload())
            msg := gmqtt.NewMessage(msg.Topic(), b, msg.Qos())
            // this line stucks somehow
            p.service.PublishService().Publish(msg)
            // return false here so the compressed payload won't be forwarded
            return false
        }

        // forward the decompressed payload
        return true
    }
}

[DrmagicE]

I test with the following code and it works as expected. You should use *Plugin instead of Plugin as the method receiver. @maddie

package main

import (
	"context"
	"log"
	"net"
	"os"
	"os/signal"
	"syscall"

	"go.uber.org/zap"

	"github.com/DrmagicE/gmqtt"
	"github.com/DrmagicE/gmqtt/pkg/packets"
)

type Plugin struct {
	s gmqtt.Server
}

func (p *Plugin) Load(service gmqtt.Server) error {
	p.s = service
	return nil
}

func (p *Plugin) Unload() error {
	return nil
}

func (p *Plugin) HookWrapper() gmqtt.HookWrapper {
	return gmqtt.HookWrapper{
		OnMsgArrivedWrapper: func(arrived gmqtt.OnMsgArrived) gmqtt.OnMsgArrived {
			return func(ctx context.Context, client gmqtt.Client, msg packets.Message) (valid bool) {
				p.s.PublishService().Publish(gmqtt.NewMessage("/abc", []byte("testing"), 1))
				return false
			}
		},
	}
}

func (p *Plugin) Name() string {
	return "testing"
}

func main() {
	// listener
	ln, err := net.Listen("tcp", ":1883")
	if err != nil {
		log.Fatalln(err.Error())
		return
	}

	//l, _ := zap.NewProduction()
	l, _ := zap.NewDevelopment()
	s := gmqtt.NewServer(
		gmqtt.WithTCPListener(ln),
		gmqtt.WithLogger(l),
	)
	p := &Plugin{
		s: s,
	}
	s.Init(gmqtt.WithPlugin(p))
	s.Run()
	signalCh := make(chan os.Signal, 1)
	signal.Notify(signalCh, os.Interrupt, syscall.SIGTERM)
	<-signalCh
	s.Stop(context.Background())
}

[maddie]

I see! That should be the problem.

Thanks a lot!