Merge branch 'master' of https://github.com/emitter-io/go

Author: kelindar

.github/FUNDING.yml

@@ -0,0 +1 @@
+github: [kelindar]

v2/emitter.go

@@ -137,9 +137,9 @@ func (c *Client) OnError(handler ErrorHandler) {
 
 // onMessage occurs when MQTT client receives a message
 func (c *Client) onMessage(_ mqtt.Client, m mqtt.Message) {
-	if c.message != nil && !strings.HasPrefix(m.Topic(), "emitter/") {
+	if !strings.HasPrefix(m.Topic(), "emitter/") {
 		handlers := c.handlers.Lookup(m.Topic())
-		if len(handlers) == 0 { // Invoke the default message handler
+		if len(handlers) == 0 && c.message != nil { // Invoke the default message handler
 			c.message(c, m)
 		}
 

v2/options.go

@@ -149,3 +149,22 @@ func WithAtMostOnce() Option {
 func WithAtLeastOnce() Option {
 	return withQos1
 }
+
+func getUTCTimestamp(input time.Time) int64 {
+	t := input
+	if zone, _ := t.Zone(); zone != "UTC" {
+		loc, _ := time.LoadLocation("UTC")
+		t = t.In(loc)
+	}
+	return t.Unix()
+}
+
+// WithFrom request messages from a point in time.
+func WithFrom(from time.Time) Option {
+	return option("from=" + strconv.FormatInt(getUTCTimestamp(from), 10))
+}
+
+// WithUntil request messages until a point in time.
+func WithUntil(until time.Time) Option {
+	return option("until=" + strconv.FormatInt(getUTCTimestamp(until), 10))
+}