-
Notifications
You must be signed in to change notification settings - Fork 198
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
New web-sockets
connector
#5170
Changes from 14 commits
0c39afd
6ab26f4
1899b85
f774a29
71523ac
162d334
2ebe548
7aa5793
42f28c7
d42c502
908ff49
48fa297
72ffd89
4d5cc99
ae1775f
fc3828f
a58411d
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -41,10 +41,16 @@ | |
# marshalled structures in block events data | ||
MarshallerType = "json" | ||
|
||
[WebSocketConnector] | ||
[HostDriverConfig] | ||
# This flag shall only be used for observer nodes | ||
Enabled = false | ||
URL = "localhost:22111" | ||
# This flag will start the WebSocket connector as server or client | ||
IsServer = false | ||
# The url of the WebSocket client/server | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I think we should clarify that this URL is used when using the driver in client mode. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This will be used in both cases because the URL is needed also for the server when it starts. |
||
URL = "127.0.0.1:22111" | ||
WithAcknowledge = true | ||
# Currently, only "json" is supported. In the future, "gogo protobuf" could also be supported | ||
MarshallerType = "json" | ||
# The number of seconds when the client will try again to send the data | ||
RetryDurationInSec = 5 | ||
BlockingAckOnError = false | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. add a descriptive comment on what this does? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. added |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,41 @@ | ||
package factory | ||
|
||
import ( | ||
"github.com/multiversx/mx-chain-communication-go/websocket/data" | ||
"github.com/multiversx/mx-chain-communication-go/websocket/factory" | ||
"github.com/multiversx/mx-chain-core-go/marshal" | ||
"github.com/multiversx/mx-chain-go/config" | ||
"github.com/multiversx/mx-chain-go/outport" | ||
"github.com/multiversx/mx-chain-go/outport/host" | ||
logger "github.com/multiversx/mx-chain-logger-go" | ||
) | ||
|
||
type ArgsHostDriverFactory struct { | ||
HostConfig config.HostDriverConfig | ||
Marshaller marshal.Marshalizer | ||
} | ||
|
||
var log = logger.GetOrCreate("outport/factory/hostdriver") | ||
|
||
func CreateHostDriver(args ArgsHostDriverFactory) (outport.Driver, error) { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. missing comments in the whole file There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. added |
||
wsHost, err := factory.CreateWebSocketHost(factory.ArgsWebSocketHost{ | ||
WebSocketConfig: data.WebSocketConfig{ | ||
URL: args.HostConfig.URL, | ||
WithAcknowledge: args.HostConfig.WithAcknowledge, | ||
IsServer: args.HostConfig.IsServer, | ||
RetryDurationInSec: args.HostConfig.RetryDurationInSec, | ||
BlockingAckOnError: args.HostConfig.BlockingAckOnError, | ||
}, | ||
Marshaller: args.Marshaller, | ||
Log: log, | ||
}) | ||
if err != nil { | ||
return nil, err | ||
} | ||
|
||
return host.NewHostDriver(host.ArgsHostDriver{ | ||
Marshaller: args.Marshaller, | ||
SenderHost: wsHost, | ||
Log: log, | ||
}) | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1 @@ | ||
package factory | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. missing unit tests There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. added |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,114 @@ | ||
package host | ||
|
||
import ( | ||
"fmt" | ||
|
||
"github.com/multiversx/mx-chain-core-go/core" | ||
"github.com/multiversx/mx-chain-core-go/core/atomic" | ||
"github.com/multiversx/mx-chain-core-go/core/check" | ||
"github.com/multiversx/mx-chain-core-go/data/outport" | ||
"github.com/multiversx/mx-chain-core-go/marshal" | ||
) | ||
|
||
// ArgsHostDriver holds the arguments needed for creating a new hostDriver | ||
type ArgsHostDriver struct { | ||
Marshaller marshal.Marshalizer | ||
SenderHost SenderHost | ||
Log core.Logger | ||
} | ||
|
||
type hostDriver struct { | ||
marshaller marshal.Marshalizer | ||
senderHost SenderHost | ||
isClosed atomic.Flag | ||
log core.Logger | ||
} | ||
|
||
func NewHostDriver(args ArgsHostDriver) (*hostDriver, error) { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. missing comment There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. added |
||
if check.IfNil(args.SenderHost) { | ||
return nil, ErrNilHost | ||
} | ||
if check.IfNil(args.Marshaller) { | ||
return nil, core.ErrNilMarshalizer | ||
} | ||
if check.IfNil(args.Log) { | ||
return nil, core.ErrNilLogger | ||
} | ||
|
||
return &hostDriver{ | ||
marshaller: args.Marshaller, | ||
senderHost: args.SenderHost, | ||
log: args.Log, | ||
isClosed: atomic.Flag{}, | ||
}, nil | ||
} | ||
|
||
func (o *hostDriver) SaveBlock(outportBlock *outport.OutportBlock) error { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. missing comment There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. added |
||
return o.handleAction(outportBlock, outport.TopicSaveBlock) | ||
} | ||
|
||
// RevertIndexedBlock will handle the action of reverting the indexed block | ||
func (o *hostDriver) RevertIndexedBlock(blockData *outport.BlockData) error { | ||
return o.handleAction(blockData, outport.TopicRevertIndexedBlock) | ||
} | ||
|
||
// SaveRoundsInfo will handle the saving of rounds | ||
func (o *hostDriver) SaveRoundsInfo(roundsInfos *outport.RoundsInfo) error { | ||
return o.handleAction(roundsInfos, outport.TopicSaveRoundsInfo) | ||
} | ||
|
||
// SaveValidatorsPubKeys will handle the saving of the validators' public keys | ||
func (o *hostDriver) SaveValidatorsPubKeys(validatorsPubKeys *outport.ValidatorsPubKeys) error { | ||
return o.handleAction(validatorsPubKeys, outport.TopicSaveValidatorsPubKeys) | ||
} | ||
|
||
// SaveValidatorsRating will handle the saving of the validators' rating | ||
func (o *hostDriver) SaveValidatorsRating(validatorsRating *outport.ValidatorsRating) error { | ||
return o.handleAction(validatorsRating, outport.TopicSaveValidatorsRating) | ||
} | ||
|
||
// SaveAccounts will handle the accounts' saving | ||
func (o *hostDriver) SaveAccounts(accounts *outport.Accounts) error { | ||
return o.handleAction(accounts, outport.TopicSaveAccounts) | ||
} | ||
|
||
// FinalizedBlock will handle the finalized block | ||
func (o *hostDriver) FinalizedBlock(finalizedBlock *outport.FinalizedBlock) error { | ||
return o.handleAction(finalizedBlock, outport.TopicFinalizedBlock) | ||
} | ||
|
||
// GetMarshaller returns the internal marshaller | ||
func (o *hostDriver) GetMarshaller() marshal.Marshalizer { | ||
return o.marshaller | ||
} | ||
|
||
func (o *hostDriver) handleAction(args interface{}, topic string) error { | ||
if o.isClosed.IsSet() { | ||
return ErrHostIsClosed | ||
} | ||
|
||
marshalledPayload, err := o.marshaller.Marshal(args) | ||
if err != nil { | ||
o.log.Error("cannot marshal block", "topic", topic, "error", err) | ||
return fmt.Errorf("%w while marshaling block for topic %s", err, topic) | ||
} | ||
|
||
err = o.senderHost.Send(marshalledPayload, topic) | ||
if err != nil { | ||
o.log.Error("cannot send on route", "topic", topic, "error", err) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Since this still can happen while closing the node, I would filter out log.Error prints only if they are not of type closed error. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. removed |
||
return fmt.Errorf("%w while sending data on route for topic %s", err, topic) | ||
} | ||
|
||
return nil | ||
} | ||
|
||
// Close will handle the closing of the outport driver web socket sender | ||
func (o *hostDriver) Close() error { | ||
o.isClosed.SetValue(true) | ||
return o.senderHost.Close() | ||
} | ||
|
||
// IsInterfaceNil returns true if there is no value under the interface | ||
func (o *hostDriver) IsInterfaceNil() bool { | ||
return o == nil | ||
} |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Maybe "ServerMode = false/true" ?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
or
Mode = "server"
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
changed in
Mode