<commit_before>package monitor

import (
	"context"
	"io"
	"sync"

	"github.com/TheThingsNetwork/ttn/api"
	pb_gateway "github.com/TheThingsNetwork/ttn/api/gateway"
	"github.com/TheThingsNetwork/ttn/api/router"
	"github.com/TheThingsNetwork/ttn/utils/errors"
	"github.com/apex/log"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/metadata"
)

// Client is a wrapper around MonitorClient
type Client struct {
	Ctx log.Interface

	client MonitorClient
	conn   *grpc.ClientConn
	addr   string

	once *sync.Once

	gateways map[string]GatewayClient
	mutex    sync.RWMutex
}

// 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) {
	cl = &Client{
		Ctx:      ctx,
		addr:     monitorAddr,
		gateways: make(map[string]GatewayClient),

		once: &sync.Once{},
	}
	return cl, cl.Open()
}

func (cl *Client) Open() (err error) {
	cl.mutex.Lock()
	defer cl.mutex.Unlock()

	return cl.open()
}
func (cl *Client) open() (err error) {
	addr := cl.addr
	ctx := cl.Ctx.WithField("addr", addr)

	defer func() {
		if err != nil {
			ctx.Warn("Failed to open monitor connection")
		} else {
			ctx.Info("Monitor connection opened")
		}
	}()

	ctx.Debug("Opening monitor connection...")

	cl.conn, err = grpc.Dial(addr, append(api.DialOptions, grpc.WithInsecure())...)
	if err != nil {
		ctx.WithError(errors.FromGRPCError(err)).Warn("Failed to establish connection to gRPC service")
		return err
	}

	cl.client = NewMonitorClient(cl.conn)
	return nil
}

// Close closes connection to the monitor
func (cl *Client) Close() (err error) {
	cl.mutex.Lock()
	defer cl.mutex.Unlock()

	return cl.close()
}
func (cl *Client) close() (err error) {
	defer func() {
		if err != nil {
			cl.Ctx.Warn("Failed to close monitor connection")
		} else {
			cl.Ctx.Info("Monitor connection closed")
		}
	}()

	for _, gtw := range cl.gateways {
		ctx := cl.Ctx.WithField("GatewayID", gtw.(*gatewayClient).id)

		ctx.Debug("Closing gateway streams...")
		err = gtw.Close()
		if err != nil {
			ctx.Warn("Failed to close gateway streams")
		}
	}

	cl.Ctx.Debug("Closing monitor connection...")
	err = cl.conn.Close()
	if err != nil {
		return err
	}

	cl.conn = nil
	return nil
}

func (cl *Client) Reopen() (err error) {
	cl.mutex.Lock()
	defer cl.mutex.Unlock()

	return cl.reopen()
}
func (cl *Client) reopen() (err error) {
	defer func() {
		if err != nil {
			cl.Ctx.Warn("Failed to reopen monitor connection")
		} else {
			cl.Ctx.Info("Monitor connection reopened")
		}
	}()

	cl.Ctx.Debug("Reopening monitor connection...")

	err = cl.close()
	if err != nil {
		return err
	}
	return cl.open()
}

func (cl *Client) IsConnected() bool {
	return cl.client != nil && cl.conn != nil
}

func (cl *Client) GatewayClient(id, token string) (gtwCl GatewayClient) {
	cl.mutex.RLock()
	gtwCl, ok := cl.gateways[id]
	cl.mutex.RUnlock()
	if !ok {
		cl.mutex.Lock()
		gtwCl = &gatewayClient{
			Ctx: cl.Ctx.WithField("GatewayID", id),

			client: cl,

			id:    id,
			token: token,
		}
		cl.gateways[id] = gtwCl
		cl.mutex.Unlock()
	}
	return gtwCl
}

type gatewayClient struct {
	client *Client

	Ctx log.Interface

	id, token string

	status struct {
		stream Monitor_GatewayStatusClient
		sync.RWMutex
	}

	uplink struct {
		stream Monitor_GatewayUplinkClient
		sync.RWMutex
	}

	downlink struct {
		stream Monitor_GatewayDownlinkClient
		sync.RWMutex
	}
}

// GatewayClient is used as the main client for Gateways to communicate with the Router
type GatewayClient interface {
	SendStatus(status *pb_gateway.Status) (err error)
	SendUplink(msg *router.UplinkMessage) (err error)
	SendDownlink(msg *router.DownlinkMessage) (err error)
	Close() (err error)
}

func (cl *gatewayClient) SendStatus(status *pb_gateway.Status) (err error) {
	cl.status.RLock()<commit_msg>monitor client: pb_gateway -> gateway<commit_after>package monitor

import (
	"context"
	"io"
	"sync"

	"github.com/TheThingsNetwork/ttn/api"
	"github.com/TheThingsNetwork/ttn/api/gateway"
	"github.com/TheThingsNetwork/ttn/api/router"
	"github.com/TheThingsNetwork/ttn/utils/errors"
	"github.com/apex/log"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/metadata"
)

// Client is a wrapper around MonitorClient
type Client struct {
	Ctx log.Interface

	client MonitorClient
	conn   *grpc.ClientConn
	addr   string

	once *sync.Once

	gateways map[string]GatewayClient
	mutex    sync.RWMutex
}

// 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) {
	cl = &Client{
		Ctx:      ctx,
		addr:     monitorAddr,
		gateways: make(map[string]GatewayClient),

		once: &sync.Once{},
	}
	return cl, cl.Open()
}

func (cl *Client) Open() (err error) {
	cl.mutex.Lock()
	defer cl.mutex.Unlock()

	return cl.open()
}
func (cl *Client) open() (err error) {
	addr := cl.addr
	ctx := cl.Ctx.WithField("addr", addr)

	defer func() {
		if err != nil {
			ctx.Warn("Failed to open monitor connection")
		} else {
			ctx.Info("Monitor connection opened")
		}
	}()

	ctx.Debug("Opening monitor connection...")

	cl.conn, err = grpc.Dial(addr, append(api.DialOptions, grpc.WithInsecure())...)
	if err != nil {
		ctx.WithError(errors.FromGRPCError(err)).Warn("Failed to establish connection to gRPC service")
		return err
	}

	cl.client = NewMonitorClient(cl.conn)
	return nil
}

// Close closes connection to the monitor
func (cl *Client) Close() (err error) {
	cl.mutex.Lock()
	defer cl.mutex.Unlock()

	return cl.close()
}
func (cl *Client) close() (err error) {
	defer func() {
		if err != nil {
			cl.Ctx.Warn("Failed to close monitor connection")
		} else {
			cl.Ctx.Info("Monitor connection closed")
		}
	}()

	for _, gtw := range cl.gateways {
		ctx := cl.Ctx.WithField("GatewayID", gtw.(*gatewayClient).id)

		ctx.Debug("Closing gateway streams...")
		err = gtw.Close()
		if err != nil {
			ctx.Warn("Failed to close gateway streams")
		}
	}

	cl.Ctx.Debug("Closing monitor connection...")
	err = cl.conn.Close()
	if err != nil {
		return err
	}

	cl.conn = nil
	return nil
}

func (cl *Client) Reopen() (err error) {
	cl.mutex.Lock()
	defer cl.mutex.Unlock()

	return cl.reopen()
}
func (cl *Client) reopen() (err error) {
	defer func() {
		if err != nil {
			cl.Ctx.Warn("Failed to reopen monitor connection")
		} else {
			cl.Ctx.Info("Monitor connection reopened")
		}
	}()

	cl.Ctx.Debug("Reopening monitor connection...")

	err = cl.close()
	if err != nil {
		return err
	}
	return cl.open()
}

func (cl *Client) IsConnected() bool {
	return cl.client != nil && cl.conn != nil
}

func (cl *Client) GatewayClient(id, token string) (gtwCl GatewayClient) {
	cl.mutex.RLock()
	gtwCl, ok := cl.gateways[id]
	cl.mutex.RUnlock()
	if !ok {
		cl.mutex.Lock()
		gtwCl = &gatewayClient{
			Ctx: cl.Ctx.WithField("GatewayID", id),

			client: cl,

			id:    id,
			token: token,
		}
		cl.gateways[id] = gtwCl
		cl.mutex.Unlock()
	}
	return gtwCl
}

type gatewayClient struct {
	client *Client

	Ctx log.Interface

	id, token string

	status struct {
		stream Monitor_GatewayStatusClient
		sync.RWMutex
	}

	uplink struct {
		stream Monitor_GatewayUplinkClient
		sync.RWMutex
	}

	downlink struct {
		stream Monitor_GatewayDownlinkClient
		sync.RWMutex
	}
}

// GatewayClient is used as the main client for Gateways to communicate with the Router
type GatewayClient interface {
	SendStatus(status *gateway.Status) (err error)
	SendUplink(msg *router.UplinkMessage) (err error)
	SendDownlink(msg *router.DownlinkMessage) (err error)
	Close() (err error)
}

func (cl *gatewayClient) SendStatus(status *gateway.Status) (err error) {
	cl.status.RLock()
{'fraction_non_alphanumeric': 0.13701953773464437, 'fraction_numerical': 0.0, 'mean_word_length': 3.720916214587101, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3787588', 'n_tokens_mistral': 2950, 'n_tokens_neox': 2821, 'n_words': 715}