mirror of
https://github.com/kaspanet/kaspad.git
synced 2025-05-21 14:26:45 +00:00
d14809694f
* [NOD-1223] Delete unused files/packages. * [NOD-1223] Move signal and limits to the os package. * [NOD-1223] Put database and dbaccess into the db package. * [NOD-1223] Fold the logs package into the logger package. * [NOD-1223] Rename domainmessage to appmessage. * [NOD-1223] Rename to/from DomainMessage to AppMessage. * [NOD-1223] Move appmessage to the app packge. * [NOD-1223] Move protocol to the app packge. * [NOD-1223] Move the network package to the infrastructure packge. * [NOD-1223] Rename cmd to executables. * [NOD-1223] Fix go.doc in the logger package.
108 lines
2.7 KiB
Go
108 lines
2.7 KiB
Go
package router
|
|
|
|
import (
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/kaspanet/kaspad/app/protocol/protocolerrors"
|
|
|
|
"github.com/kaspanet/kaspad/app/appmessage"
|
|
"github.com/pkg/errors"
|
|
)
|
|
|
|
const (
|
|
// DefaultMaxMessages is the default capacity for a route with a capacity defined
|
|
DefaultMaxMessages = 100
|
|
)
|
|
|
|
var (
|
|
// ErrTimeout signifies that one of the router functions had a timeout.
|
|
ErrTimeout = protocolerrors.New(false, "timeout expired")
|
|
|
|
// ErrRouteClosed indicates that a route was closed while reading/writing.
|
|
ErrRouteClosed = errors.New("route is closed")
|
|
)
|
|
|
|
// onCapacityReachedHandler is a function that is to be
|
|
// called when a route reaches capacity.
|
|
type onCapacityReachedHandler func()
|
|
|
|
// Route represents an incoming or outgoing Router route
|
|
type Route struct {
|
|
channel chan appmessage.Message
|
|
// closed and closeLock are used to protect us from writing to a closed channel
|
|
// reads use the channel's built-in mechanism to check if the channel is closed
|
|
closed bool
|
|
closeLock sync.Mutex
|
|
|
|
onCapacityReachedHandler onCapacityReachedHandler
|
|
}
|
|
|
|
// NewRoute create a new Route
|
|
func NewRoute() *Route {
|
|
return newRouteWithCapacity(DefaultMaxMessages)
|
|
}
|
|
|
|
func newRouteWithCapacity(capacity int) *Route {
|
|
return &Route{
|
|
channel: make(chan appmessage.Message, capacity),
|
|
closed: false,
|
|
}
|
|
}
|
|
|
|
// Enqueue enqueues a message to the Route
|
|
func (r *Route) Enqueue(message appmessage.Message) error {
|
|
r.closeLock.Lock()
|
|
defer r.closeLock.Unlock()
|
|
|
|
if r.closed {
|
|
return errors.WithStack(ErrRouteClosed)
|
|
}
|
|
if len(r.channel) == DefaultMaxMessages {
|
|
r.onCapacityReachedHandler()
|
|
}
|
|
r.channel <- message
|
|
return nil
|
|
}
|
|
|
|
// Dequeue dequeues a message from the Route
|
|
func (r *Route) Dequeue() (appmessage.Message, error) {
|
|
message, isOpen := <-r.channel
|
|
if !isOpen {
|
|
return nil, errors.WithStack(ErrRouteClosed)
|
|
}
|
|
return message, nil
|
|
}
|
|
|
|
// DequeueWithTimeout attempts to dequeue a message from the Route
|
|
// and returns an error if the given timeout expires first.
|
|
func (r *Route) DequeueWithTimeout(timeout time.Duration) (appmessage.Message, error) {
|
|
select {
|
|
case <-time.After(timeout):
|
|
return nil, errors.Wrapf(ErrTimeout, "got timeout after %s", timeout)
|
|
case message, isOpen := <-r.channel:
|
|
if !isOpen {
|
|
return nil, errors.WithStack(ErrRouteClosed)
|
|
}
|
|
return message, nil
|
|
}
|
|
}
|
|
|
|
func (r *Route) setOnCapacityReachedHandler(onCapacityReachedHandler onCapacityReachedHandler) {
|
|
r.onCapacityReachedHandler = onCapacityReachedHandler
|
|
}
|
|
|
|
// Close closes this route
|
|
func (r *Route) Close() {
|
|
r.closeLock.Lock()
|
|
defer r.closeLock.Unlock()
|
|
|
|
r.closed = true
|
|
close(r.channel)
|
|
}
|
|
|
|
// IsEmpty returns true if the route doesn't have any pending messages
|
|
func (r *Route) IsEmpty() bool {
|
|
return len(r.channel) == 0
|
|
}
|