Merge pull request #436 from TheThingsNetwork/develop

v2.3.0

amqp/downlink.go

@@ -42,7 +42,6 @@ func (s *DefaultSubscriber) SubscribeDeviceDownlink(appID, devID string, handler
 			}
 			handler(s, dataDown.AppID, dataDown.DevID, *dataDown)
 			delivery.Ack(false)
-			break
 		}
 	}()
 

amqp/utils_test.go

@@ -6,11 +6,10 @@ package amqp
 import (
 	"testing"
 
-	"github.com/TheThingsNetwork/go-utils/log"
-	"github.com/TheThingsNetwork/go-utils/log/apex"
+	ttnlog "github.com/TheThingsNetwork/go-utils/log"
 	tt "github.com/TheThingsNetwork/ttn/utils/testing"
 )
 
-func getLogger(t *testing.T, tag string) log.Interface {
-	return apex.Wrap(tt.GetLogger(t, tag))
+func getLogger(t *testing.T, tag string) ttnlog.Interface {
+	return tt.GetLogger(t, tag)
 }

api/broker/communication_test.go

@@ -11,7 +11,6 @@ import (
 	"time"
 
 	"github.com/TheThingsNetwork/go-utils/log"
-	"github.com/TheThingsNetwork/go-utils/log/apex"
 	"github.com/TheThingsNetwork/ttn/api"
 	. "github.com/TheThingsNetwork/ttn/utils/testing"
 	. "github.com/smartystreets/assertions"
@@ -51,7 +50,7 @@ func TestHandlerBrokerCommunication(t *testing.T) {
 	a := New(t)
 
 	ctx := GetLogger(t, "TestHandlerBrokerCommunication")
-	log.Set(apex.Wrap(ctx))
+	log.Set(ctx)
 
 	brk := newTestBroker()
 	rand.Seed(time.Now().UnixNano())
@@ -145,7 +144,7 @@ func TestRouterBrokerCommunication(t *testing.T) {
 	a := New(t)
 
 	ctx := GetLogger(t, "TestRouterBrokerCommunication")
-	log.Set(apex.Wrap(ctx))
+	log.Set(ctx)
 
 	brk := newTestBroker()
 	rand.Seed(time.Now().UnixNano())

api/context.go

@@ -28,3 +28,19 @@ func IDFromContext(ctx context.Context) (token string, err error) {
 	}
 	return IDFromMetadata(md)
 }
+
+func LimitAndOffsetFromContext(ctx context.Context) (limit, offset int, err error) {
+	md, err := MetadataFromContext(ctx)
+	if err != nil {
+		return 0, 0, err
+	}
+	limit, err = LimitFromMetadata(md)
+	if err != nil {
+		return 0, 0, err
+	}
+	offset, err = OffsetFromMetadata(md)
+	if err != nil {
+		return 0, 0, err
+	}
+	return limit, offset, nil
+}

api/handler/manager_client.go

@@ -7,6 +7,7 @@ import (
 	"encoding/json"
 	"os"
 	"os/user"
+	"strconv"
 	"sync"
 
 	"github.com/TheThingsNetwork/ttn/api/protocol/lorawan"
@@ -60,7 +61,8 @@ func (h *ManagerClient) UpdateAccessToken(accessToken string) {
 	h.accessToken = accessToken
 }
 
-func (h *ManagerClient) getContext() context.Context {
+// GetContext returns a new context with authentication
+func (h *ManagerClient) GetContext() context.Context {
 	h.RLock()
 	defer h.RUnlock()
 	md := metadata.Pairs(
@@ -70,9 +72,22 @@ func (h *ManagerClient) getContext() context.Context {
 	return metadata.NewContext(context.Background(), md)
 }
 
+// GetContext returns a new context with authentication, plus limit and offset for pagination
+func (h *ManagerClient) GetContextWithLimitAndOffset(limit, offset int) context.Context {
+	h.RLock()
+	defer h.RUnlock()
+	md := metadata.Pairs(
+		"id", h.id,
+		"token", h.accessToken,
+		"limit", strconv.Itoa(limit),
+		"offset", strconv.Itoa(offset),
+	)
+	return metadata.NewContext(context.Background(), md)
+}
+
 // GetApplication retrieves an application from the Handler
 func (h *ManagerClient) GetApplication(appID string) (*Application, error) {
-	res, err := h.applicationManagerClient.GetApplication(h.getContext(), &ApplicationIdentifier{AppId: appID})
+	res, err := h.applicationManagerClient.GetApplication(h.GetContext(), &ApplicationIdentifier{AppId: appID})
 	if err != nil {
 		return nil, errors.Wrap(errors.FromGRPCError(err), "Could not get application from Handler")
 	}
@@ -81,25 +96,25 @@ func (h *ManagerClient) GetApplication(appID string) (*Application, error) {
 
 // SetApplication sets an application on the Handler
 func (h *ManagerClient) SetApplication(in *Application) error {
-	_, err := h.applicationManagerClient.SetApplication(h.getContext(), in)
+	_, err := h.applicationManagerClient.SetApplication(h.GetContext(), in)
 	return errors.Wrap(errors.FromGRPCError(err), "Could not set application on Handler")
 }
 
 // RegisterApplication registers an application on the Handler
 func (h *ManagerClient) RegisterApplication(appID string) error {
-	_, err := h.applicationManagerClient.RegisterApplication(h.getContext(), &ApplicationIdentifier{AppId: appID})
+	_, err := h.applicationManagerClient.RegisterApplication(h.GetContext(), &ApplicationIdentifier{AppId: appID})
 	return errors.Wrap(errors.FromGRPCError(err), "Could not register application on Handler")
 }
 
 // DeleteApplication deletes an application and all its devices from the Handler
 func (h *ManagerClient) DeleteApplication(appID string) error {
-	_, err := h.applicationManagerClient.DeleteApplication(h.getContext(), &ApplicationIdentifier{AppId: appID})
+	_, err := h.applicationManagerClient.DeleteApplication(h.GetContext(), &ApplicationIdentifier{AppId: appID})
 	return errors.Wrap(errors.FromGRPCError(err), "Could not delete application from Handler")
 }
 
 // GetDevice retrieves a device from the Handler
 func (h *ManagerClient) GetDevice(appID string, devID string) (*Device, error) {
-	res, err := h.applicationManagerClient.GetDevice(h.getContext(), &DeviceIdentifier{AppId: appID, DevId: devID})
+	res, err := h.applicationManagerClient.GetDevice(h.GetContext(), &DeviceIdentifier{AppId: appID, DevId: devID})
 	if err != nil {
 		return nil, errors.Wrap(errors.FromGRPCError(err), "Could not get device from Handler")
 	}
@@ -108,19 +123,20 @@ func (h *ManagerClient) GetDevice(appID string, devID string) (*Device, error) {
 
 // SetDevice sets a device on the Handler
 func (h *ManagerClient) SetDevice(in *Device) error {
-	_, err := h.applicationManagerClient.SetDevice(h.getContext(), in)
+	_, err := h.applicationManagerClient.SetDevice(h.GetContext(), in)
 	return errors.Wrap(errors.FromGRPCError(err), "Could not set device on Handler")
 }
 
 // DeleteDevice deletes a device from the Handler
 func (h *ManagerClient) DeleteDevice(appID string, devID string) error {
-	_, err := h.applicationManagerClient.DeleteDevice(h.getContext(), &DeviceIdentifier{AppId: appID, DevId: devID})
+	_, err := h.applicationManagerClient.DeleteDevice(h.GetContext(), &DeviceIdentifier{AppId: appID, DevId: devID})
 	return errors.Wrap(errors.FromGRPCError(err), "Could not delete device from Handler")
 }
 
-// GetDevicesForApplication retrieves all devices for an application from the Handler
-func (h *ManagerClient) GetDevicesForApplication(appID string) (devices []*Device, err error) {
-	res, err := h.applicationManagerClient.GetDevicesForApplication(h.getContext(), &ApplicationIdentifier{AppId: appID})
+// GetDevicesForApplication retrieves all devices for an application from the Handler.
+// Pass a limit to indicate the maximum number of results you want to receive, and the offset to indicate how many results should be skipped.
+func (h *ManagerClient) GetDevicesForApplication(appID string, limit, offset int) (devices []*Device, err error) {
+	res, err := h.applicationManagerClient.GetDevicesForApplication(h.GetContextWithLimitAndOffset(limit, offset), &ApplicationIdentifier{AppId: appID})
 	if err != nil {
 		return nil, errors.Wrap(errors.FromGRPCError(err), "Could not get devices for application from Handler")
 	}
@@ -133,7 +149,7 @@ func (h *ManagerClient) GetDevicesForApplication(appID string) (devices []*Devic
 // GetDevAddr requests a random device address with the given constraints
 func (h *ManagerClient) GetDevAddr(constraints ...string) (types.DevAddr, error) {
 	devAddrManager := lorawan.NewDevAddrManagerClient(h.conn)
-	resp, err := devAddrManager.GetDevAddr(h.getContext(), &lorawan.DevAddrRequest{
+	resp, err := devAddrManager.GetDevAddr(h.GetContext(), &lorawan.DevAddrRequest{
 		Usage: constraints,
 	})
 	if err != nil {
@@ -145,7 +161,7 @@ func (h *ManagerClient) GetDevAddr(constraints ...string) (types.DevAddr, error)
 // DryUplink transforms the uplink payload with the payload functions provided
 // in the app..
 func (h *ManagerClient) DryUplink(payload []byte, app *Application, port uint32) (*DryUplinkResult, error) {
-	res, err := h.applicationManagerClient.DryUplink(h.getContext(), &DryUplinkMessage{
+	res, err := h.applicationManagerClient.DryUplink(h.GetContext(), &DryUplinkMessage{
 		App:     app,
 		Payload: payload,
 		Port:    port,
@@ -159,7 +175,7 @@ func (h *ManagerClient) DryUplink(payload []byte, app *Application, port uint32)
 // DryDownlinkWithPayload transforms the downlink payload with the payload functions
 // provided in app.
 func (h *ManagerClient) DryDownlinkWithPayload(payload []byte, app *Application, port uint32) (*DryDownlinkResult, error) {
-	res, err := h.applicationManagerClient.DryDownlink(h.getContext(), &DryDownlinkMessage{
+	res, err := h.applicationManagerClient.DryDownlink(h.GetContext(), &DryDownlinkMessage{
 		App:     app,
 		Payload: payload,
 		Port:    port,
@@ -178,7 +194,7 @@ func (h *ManagerClient) DryDownlinkWithFields(fields map[string]interface{}, app
 		return nil, err
 	}
 
-	res, err := h.applicationManagerClient.DryDownlink(h.getContext(), &DryDownlinkMessage{
+	res, err := h.applicationManagerClient.DryDownlink(h.GetContext(), &DryDownlinkMessage{
 		App:    app,
 		Fields: string(marshalled),
 		Port:   port,

api/monitor/client.go

@@ -6,12 +6,12 @@ package monitor
 import (
 	"sync"
 
+	ttnlog "github.com/TheThingsNetwork/go-utils/log"
 	"github.com/TheThingsNetwork/ttn/api"
 	"github.com/TheThingsNetwork/ttn/api/broker"
 	"github.com/TheThingsNetwork/ttn/api/gateway"
 	"github.com/TheThingsNetwork/ttn/api/router"
 	"github.com/TheThingsNetwork/ttn/utils/errors"
-	"github.com/apex/log"
 	"golang.org/x/net/context" // See https://github.com/grpc/grpc-go/issues/711"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/metadata"
@@ -22,7 +22,7 @@ const BufferSize = 10
 
 // Client is a wrapper around MonitorClient
 type Client struct {
-	Ctx log.Interface
+	Ctx ttnlog.Interface
 
 	client MonitorClient
 	conn   *grpc.ClientConn
@@ -36,7 +36,7 @@ type Client struct {
 
 // NewClient is a wrapper for NewMonitorClient, initializes
 // connection to MonitorServer on monitorAddr with default gRPC options
-func NewClient(ctx log.Interface, monitorAddr string) (cl *Client, err error) {
+func NewClient(ctx ttnlog.Interface, monitorAddr string) (cl *Client, err error) {
 	cl = &Client{
 		Ctx:      ctx,
 		addr:     monitorAddr,
@@ -171,7 +171,7 @@ type gatewayClient struct {
 
 	client *Client
 
-	Ctx log.Interface
+	Ctx ttnlog.Interface
 
 	id, token string
 
@@ -242,7 +242,7 @@ type brokerClient struct {
 
 	client *Client
 
-	Ctx log.Interface
+	Ctx ttnlog.Interface
 
 	uplink struct {
 		init   sync.Once

api/monitor/registry.go

@@ -6,7 +6,7 @@ package monitor
 import (
 	"sync"
 
-	"github.com/apex/log"
+	ttnlog "github.com/TheThingsNetwork/go-utils/log"
 )
 
 // Registry encapsulates dealing with monitor servers that might be down during startup.
@@ -25,7 +25,7 @@ type Registry interface {
 }
 
 // NewRegistry creates a monitor client registry.
-func NewRegistry(ctx log.Interface) Registry {
+func NewRegistry(ctx ttnlog.Interface) Registry {
 	return &registry{
 		ctx:              ctx,
 		monitorClients:   make(map[string]*Client),
@@ -37,12 +37,12 @@ func NewRegistry(ctx log.Interface) Registry {
 }
 
 type registry struct {
-	ctx              log.Interface
+	ctx              ttnlog.Interface
 	monitorClients   map[string]*Client
 	brokerClients    []BrokerClient
 	gatewayClients   map[string][]GatewayClient
 	gatewayTokens    map[string]string
-	newMonitorClient func(ctx log.Interface, addr string) (*Client, error)
+	newMonitorClient func(ctx ttnlog.Interface, addr string) (*Client, error)
 	sync.RWMutex
 }
 

api/monitor/registry_test.go

@@ -7,8 +7,8 @@ import (
 	"errors"
 	"testing"
 
+	ttnlog "github.com/TheThingsNetwork/go-utils/log"
 	. "github.com/TheThingsNetwork/ttn/utils/testing"
-	"github.com/apex/log"
 	. "github.com/smartystreets/assertions"
 )
 
@@ -140,14 +140,14 @@ func TestGatewayClients(t *testing.T) {
 	})
 }
 
-var returnClient = func(ctx log.Interface, addr string) (*Client, error) {
+var returnClient = func(ctx ttnlog.Interface, addr string) (*Client, error) {
 	return &Client{
 		Ctx:          ctx,
 		BrokerClient: &brokerClient{},
 		gateways:     make(map[string]GatewayClient),
 	}, nil
 }
 
-var returnError = func(ctx log.Interface, addr string) (*Client, error) {
+var returnError = func(ctx ttnlog.Interface, addr string) (*Client, error) {
 	return nil, errors.New("")
 }

api/router/communication_test.go

@@ -11,7 +11,6 @@ import (
 	"time"
 
 	"github.com/TheThingsNetwork/go-utils/log"
-	"github.com/TheThingsNetwork/go-utils/log/apex"
 	"github.com/TheThingsNetwork/ttn/api"
 	"github.com/TheThingsNetwork/ttn/api/gateway"
 	"github.com/TheThingsNetwork/ttn/api/protocol"
@@ -52,7 +51,7 @@ func TestRouterCommunication(t *testing.T) {
 	a := New(t)
 
 	ctx := GetLogger(t, "TestRouterCommunication")
-	log.Set(apex.Wrap(ctx))
+	log.Set(ctx)
 
 	rtr := newTestRouter()
 	rand.Seed(time.Now().UnixNano())

cmd/broker.go

@@ -14,9 +14,9 @@ import (
 
 	"google.golang.org/grpc"
 
+	ttnlog "github.com/TheThingsNetwork/go-utils/log"
 	"github.com/TheThingsNetwork/ttn/core/broker"
 	"github.com/TheThingsNetwork/ttn/core/component"
-	"github.com/apex/log"
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
 )
@@ -27,7 +27,7 @@ var brokerCmd = &cobra.Command{
 	Short: "The Things Network broker",
 	Long:  ``,
 	PreRun: func(cmd *cobra.Command, args []string) {
-		ctx.WithFields(log.Fields{
+		ctx.WithFields(ttnlog.Fields{
 			"Server":             fmt.Sprintf("%s:%d", viper.GetString("broker.server-address"), viper.GetInt("broker.server-port")),
 			"Announce":           fmt.Sprintf("%s:%d", viper.GetString("broker.server-address-announce"), viper.GetInt("broker.server-port")),
 			"NetworkServer":      viper.GetString("broker.networkserver-address"),
@@ -38,7 +38,7 @@ var brokerCmd = &cobra.Command{
 		ctx.Info("Starting")
 
 		// Component
-		component, err := component.New(ctx, "broker", fmt.Sprintf("%s:%d", viper.GetString("broker.server-address-announce"), viper.GetInt("broker.server-port")))
+		component, err := component.New(ttnlog.Get(), "broker", fmt.Sprintf("%s:%d", viper.GetString("broker.server-address-announce"), viper.GetInt("broker.server-port")))
 		if err != nil {
 			ctx.WithError(err).Fatal("Could not initialize component")
 		}

cmd/discovery.go

@@ -12,12 +12,12 @@ import (
 	"os/signal"
 	"syscall"
 
+	ttnlog "github.com/TheThingsNetwork/go-utils/log"
 	pb "github.com/TheThingsNetwork/ttn/api/discovery"
 	"github.com/TheThingsNetwork/ttn/core/component"
 	"github.com/TheThingsNetwork/ttn/core/discovery"
 	"github.com/TheThingsNetwork/ttn/core/discovery/announcement"
 	"github.com/TheThingsNetwork/ttn/core/proxy"
-	"github.com/apex/log"
 	"github.com/grpc-ecosystem/grpc-gateway/runtime"
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
@@ -31,7 +31,7 @@ var discoveryCmd = &cobra.Command{
 	Short: "The Things Network discovery",
 	Long:  ``,
 	PreRun: func(cmd *cobra.Command, args []string) {
-		ctx.WithFields(log.Fields{
+		ctx.WithFields(ttnlog.Fields{
 			"Server":     fmt.Sprintf("%s:%d", viper.GetString("discovery.server-address"), viper.GetInt("discovery.server-port")),
 			"HTTP Proxy": fmt.Sprintf("%s:%d", viper.GetString("discovery.http-address"), viper.GetInt("discovery.http-port")),
 			"Database":   fmt.Sprintf("%s/%d", viper.GetString("discovery.redis-address"), viper.GetInt("discovery.redis-db")),
@@ -50,7 +50,7 @@ var discoveryCmd = &cobra.Command{
 		connectRedis(client)
 
 		// Component
-		component, err := component.New(ctx, "discovery", fmt.Sprintf("%s:%d", "localhost", viper.GetInt("discovery.server-port")))
+		component, err := component.New(ttnlog.Get(), "discovery", fmt.Sprintf("%s:%d", "localhost", viper.GetInt("discovery.server-port")))
 		if err != nil {
 			ctx.WithError(err).Fatal("Could not initialize component")
 		}
@@ -89,6 +89,7 @@ var discoveryCmd = &cobra.Command{
 			pb.RegisterDiscoveryHandler(netCtx, mux, proxyConn)
 
 			prxy := proxy.WithLogger(mux, ctx)
+			prxy = proxy.WithPagination(prxy)
 
 			go func() {
 				err := http.ListenAndServe(