Wrapping callbacks dynamically with a Goroutine using reflect
Tags: nats, reflect, go
Published: 2019-04-04 06:09:16

---
title: Wrapping callbacks dynamically with a Goroutine using reflect
published: true
description: Using reflect to dynamically attach a callback to a Goroutine
tags: nats, reflect, golang
---

Recently a NATS user was looking for a generic way to be able to
[increase the concurrency of the async NATS handler subscriptions](https://github.com/nats-io/go-nats/pull/451) which
by default process the messages sequentially in order. This means that
if a message takes very long to be processed, then the next message
will be processed with some delay even if the server has already
delivered it to the client: 

```go
nc, err := nats.Connect("demo.nats.io")
if err != nil {
        log.Fatalf("Failed to create a connection: %v\n", err)
}

nc.Subscribe("foo", func(m *nats.Msg) {
        log.Printf("[Received on '>'  ] %+v", string(msg.Data))
        time.Sleep(1 * time.Second) // Blocks each message for 1 second
})
```

An easy way to work around the head of line blocking, would be to
simply add a Goroutine:

```go
nc.Subscribe("foo", func(m *nats.Msg) {
	log.Printf("[Received on '>'  ] %+v", string(msg.Data))
        go func(){
		// ... Processing ...
		time.Sleep(1 * time.Second) 
	}()
})
```

This works for regular NATS connections and subscriptions so it is
often the approach I suggest.  There is one exception to this though,
which is when using NATS encoded connections, the type of the function
can be of a multiple number of types, with the NATS client internally
handling the serialization/deserialization of the payload:

```go
package main

import (
	"log"
	"time"

	"github.com/nats-io/go-nats"
)

func init() {
	log.SetFlags(log.LstdFlags | log.Lmicroseconds)
}

func main() {
	nc, err := nats.Connect("demo.nats.io")
	if err != nil {
		log.Fatalf("Failed to create a connection: %v\n", err)
	}
	ec, err := nats.NewEncodedConn(nc, nats.JSON_ENCODER)
	if err != nil {
		log.Fatalf("Failed to create an encoded connection: %v\n", err)
	}
	defer ec.Close()

	type Event struct {
		Message string `json:"message"`
	}

	ec.Subscribe(">", func(b []byte) {
		log.Printf("[Received on '>'  ] %+v", string(b))
		time.Sleep(1 * time.Second)
	})

	ec.Subscribe("foo", func(s string) {
		log.Printf("[Received on 'foo'] %+v", s)
		time.Sleep(1 * time.Second)
	})

	ec.Subscribe("bar", func(e *Event) {
		log.Printf("[Received on 'bar'] %+v", e)
		time.Sleep(1 * time.Second)
	})

	payload := []byte("Hello World!")
	for i := 0; i < 3; i++ {
		ec.Publish("foo", payload)
		ec.Publish("bar", &Event{Message: string(payload)})
	}
	select {
	case <-time.After(2 * time.Second):
	}

}
```

Now the simple workaround for the Goroutine does not work since would
have to implement it for a multiple number of function types. This is
because internally the NATS client uses `reflect` to find out how to
properly call the callback using the decoded payload.

In order to workaround this issue, I [dug my way out](https://www.youtube.com/watch?v=_znElks13UA)
using more reflection to dynamically rewrap a function so that it is
ran in its own Goroutine:

```go
type subscriber interface {
	Subscribe(subject string, cb nats.Handler) (*nats.Subscription, error)
	QueueSubscribe(subject, group string, cb nats.Handler) (*nats.Subscription, error)
}

func Subscribe(ec subscriber, subject string, cb nats.Handler) (*nats.Subscription, error) {
	parallelize := func(f, fptr interface{}) {
		// Original function.
		oV := reflect.ValueOf(f)

		// Pointer to the function that will be replaced.
		ffn := reflect.ValueOf(fptr).Interface()
		fn := reflect.ValueOf(ffn).Elem()

		// Type of original function used to create new one
		// that will replace the original.
		typ := reflect.TypeOf(f)

		// Rewrap original function.
		handler := func(in []reflect.Value) []reflect.Value {
			go oV.Call(in)
			return nil
		}
		v := reflect.MakeFunc(typ, handler)
		fn.Set(v)
	}
	parallelize(cb, &cb)
	return ec.Subscribe(subject, cb)
}
```

Now a user can use this helper function that has a similar API as the
original NATS client subscribe, though internally it is handling the
messages each in its own Goroutine:

```go
Subscribe(ec, ">", func(b []byte) {
        log.Printf("[Received on '>'  ] %+v", string(b))
        time.Sleep(1 * time.Second)
})

Subscribe(ec, "foo", func(s string) {
        log.Printf("[Received on 'foo'] %+v", s)
        time.Sleep(1 * time.Second)
})

Subscribe(ec, "bar", func(e *Event) {
        log.Printf("[Received on 'bar'] %+v", e)
        time.Sleep(1 * time.Second)
})
```

Full example below:

```go
package main

import (
	"log"
	"reflect"
	"time"

	"github.com/nats-io/go-nats"
)

type subscriber interface {
	Subscribe(subject string, cb nats.Handler) (*nats.Subscription, error)
	QueueSubscribe(subject, group string, cb nats.Handler) (*nats.Subscription, error)
}

func Subscribe(ec subscriber, subject string, cb nats.Handler) (*nats.Subscription, error) {
	parallelize := func(f, fptr interface{}) {
		// Original function.
		oV := reflect.ValueOf(f)

		// Pointer to the function that will be replaced.
		ffn := reflect.ValueOf(fptr).Interface()
		fn := reflect.ValueOf(ffn).Elem()

		// Type of original function used to create new one
		// that will replace the original.
		typ := reflect.TypeOf(f)

		// Rewrap original function.
		handler := func(in []reflect.Value) []reflect.Value {
			go oV.Call(in)
			return nil
		}
		v := reflect.MakeFunc(typ, handler)
		fn.Set(v)
	}
	parallelize(cb, &cb)
	return ec.Subscribe(subject, cb)
}

func init() {
	log.SetFlags(log.LstdFlags | log.Lmicroseconds)
}

func main() {
	nc, err := nats.Connect("demo.nats.io")
	if err != nil {
		log.Fatalf("Failed to create a connection: %v\n", err)
	}
	ec, err := nats.NewEncodedConn(nc, nats.JSON_ENCODER)
	if err != nil {
		log.Fatalf("Failed to create an encoded connection: %v\n", err)
	}
	defer ec.Close()

	type Event struct {
		Message string `json:"message"`
	}

	Subscribe(ec, ">", func(b []byte) {
		log.Printf("[Received on '>'  ] %+v", string(b))
		time.Sleep(1 * time.Second)
	})

	Subscribe(ec, "foo", func(s string) {
		log.Printf("[Received on 'foo'] %+v", s)
		time.Sleep(1 * time.Second)
	})

	Subscribe(ec, "bar", func(e *Event) {
		log.Printf("[Received on 'bar'] %+v", e)
		time.Sleep(1 * time.Second)
	})

	testBytes := []byte("Hello World!")
	ec.Publish("foo", testBytes)
	ec.Publish("foo", testBytes)
	ec.Publish("foo", testBytes)
	ec.Publish("bar", &Event{Message: string(testBytes)})
	ec.Publish("bar", &Event{Message: string(testBytes)})
	ec.Publish("bar", &Event{Message: string(testBytes)})
	ec.PublishRequest("bar", "inbox", &Event{Message: string(testBytes)})
	select {
	case <-time.After(2 * time.Second):
	}

}
```