Errandboi is a scheduler that publishes given events to NATS and EMQX.

  • By Snapp Cab Incubators
  • Last update: Apr 19, 2022
  • Comments: 0

Errandboi

Errandboi is a scheduler that publishes given events to Nats and EMQX. It receives events via http request and caches them in Redis until their release time, then they get published to either EMQX or Nats or both. After events get published, their status changes to done.

Instructions

Lint and Run

  • run docker containers, lint and run app: make all

Build

  • build app locally: make run build
  • build docker image locally: make run docker-build

Run

  • run services: make up
  • run docker containers: make run

Scheduling and publish structure

scheduler

   ticker := time.NewTicker(d)  
  
   go func() {  
      for {  
         select {  
         case <-ticker.C:  
            sch.Publisher.GetEvents()  
            sch.Publisher.Work()  
         case <-sch.Stop:  
            sch.Publisher.Cancel()  
            sch.Publisher.Wp = workerpool.New(sch.Publisher.WorkerSize)  
  
            ticker.Stop()  
  
            return  
  }  
      }  
   }()

Nats

*creating stream

func (n *Nats) CreateStream() error {  
   stream, _ := n.JSCtx.StreamInfo(ChannelName)  
  
   if stream == nil {  
      in, err2 := n.JSCtx.AddStream(&nats.StreamConfig{  
         Name:     ChannelName,  
  Subjects: []string{SubjectName},  
  MaxAge:   0,  
  Storage:  nats.FileStorage,  
  })  
      if err2 != nil {  
         return fmt.Errorf("cannot create stream %w", err2)  
      }  
  
      stream = in  
   }  
  
   n.Logger.Info("events stream", zap.Any("stream", stream))  
  
   return nil  
}
  • publish
func (pb *Publisher) publishEventNats(event Event) {  
   t := natsp.ChannelName + "." + event.Topic  
   if _, err := pb.Nats.JSCtx.Publish(t, []byte(event.Payload)); err != nil {  
      pb.logger.Error("failed to publish event", zap.String("payload", event.Payload),  
  zap.String("topic", event.Topic), zap.Error(err))  
   }  
  
   pb.logger.Info("message published to nats", zap.String("payload", event.Payload))  
}

EMQX

  • publish
func (pb *Publisher) publishEventEMQ(event Event) {  
   if token := pb.Mqtt.Client.Subscribe(event.Topic, 0, nil); token.Wait() && token.Error() != nil {  
      fmt.Println(token.Error())  
   }  
  
   token := pb.Mqtt.Client.Publish(event.Topic, 0, false, event.Payload)  
   pb.logger.Info("message published to emq: ", zap.String("payload", event.Payload))  
   token.Wait()  
}

Example

Posting events

  • request
curl --request POST '{{baseURL}}/events' \
--data-raw '{
	"type": [ "nats","emqx"],
	"events": [
		{
		"description": "This is an event description",
		"delay": "10s",
		"topic": "topic1",
		"payload": "hello"
		}
	]
}'
  • response
{
	"id": "62495362f8358c7b9d83ff4e"
}

Check event status

  • request
curl --request GET '{{baseURL}}/events/{{eventId}}/status'
  • response
{
	"status": "done",
	"events": [
		{
		"description": "This is an event description",
		"publish_date":"2022-04-11T13:31:28+04:30",
		"status": "Done"
		}
	]
}

Download

errandboi.zip