Skip to content
Snippets Groups Projects
Commit 1d255384 authored by Jonah Husson's avatar Jonah Husson
Browse files

Remove connection status & callback code

parent 7ab227bc
No related branches found
No related tags found
No related merge requests found
Show whitespace changes
Inline Side-by-side
api/client.go
+ 2
− 37
View file @ 1d255384
......@@ -154,8 +154,7 @@ func requestNdf(cl *Client) error {
// If none is provided, a default storage using OS file access
// is created
// returns a new Client object, and an error if it fails
func NewClient(s globals.Storage, locA, locB string, ndfJSON *ndf.NetworkDefinition,
callback io.ConnectionStatusCallback) (*Client, error) {
func NewClient(s globals.Storage, locA, locB string, ndfJSON *ndf.NetworkDefinition) (*Client, error) {
var store globals.Storage
if s == nil {
globals.Log.INFO.Printf("No storage provided," +
......@@ -175,7 +174,7 @@ func NewClient(s globals.Storage, locA, locB string, ndfJSON *ndf.NetworkDefinit
cl := new(Client)
cl.storage = store
cl.commManager = io.NewCommManager(ndfJSON, callback)
cl.commManager = io.NewCommManager(ndfJSON)
cl.ndf = ndfJSON
//build the topology
nodeIDs := make([]*id.Node, len(cl.ndf.Nodes))
......@@ -277,10 +276,6 @@ const SaltSize = 256
func (cl *Client) RegisterWithPermissioning(preCan bool, registrationCode, nick, email,
password string, privateKeyRSA *rsa.PrivateKey) (*id.User, error) {
if !preCan && cl.commManager.GetConnectionStatus() != io.Online {
return nil, errors.New("Cannot register when disconnected from the network")
}
var err error
var u *user.User
var UID *id.User
......@@ -422,11 +417,6 @@ func (cl *Client) RegisterWithUDB(timeout time.Duration) error {
"state is not PermissioningComplete")
}
status := cl.commManager.GetConnectionStatus()
if status == io.Connecting || status == io.Offline {
return errors.New("ERROR: could not RegisterWithUDB - connection is either offline or connecting")
}
email := cl.session.GetCurrentUser().Email
var err error
......@@ -642,11 +632,6 @@ func (cl *Client) Login(password string) (string, error) {
// Logs in user and sets session on client object
// returns the nickname or error if login fails
func (cl *Client) StartMessageReceiver() error {
status := cl.commManager.GetConnectionStatus()
if status == io.Connecting || status == io.Offline {
return errors.New("ERROR: could not StartMessageReceiver - connection is either offline or connecting")
}
// Initialize UDB and nickname "bot" stuff here
bots.InitBots(cl.session, cl.commManager, cl.topology, id.NewUserFromBytes(cl.ndf.UDB.ID))
// Initialize Rekey listeners
......@@ -663,11 +648,6 @@ func (cl *Client) StartMessageReceiver() error {
// Send prepares and sends a message to the cMix network
// FIXME: We need to think through the message interface part.
func (cl *Client) Send(message parse.MessageInterface) error {
status := cl.commManager.GetConnectionStatus()
if status == io.Connecting || status == io.Offline {
return errors.New("Could not Send - connection is either offline or connecting")
}
// FIXME: There should (at least) be a version of this that takes a byte array
recipientID := message.GetRecipient()
cryptoType := message.GetCryptoType()
......@@ -710,10 +690,6 @@ func (cl *Client) GetKeyParams() *keyStore.KeyParams {
return cl.session.GetKeyStore().GetKeyParams()
}
func (cl *Client) GetNetworkStatus() uint32 {
return cl.commManager.GetConnectionStatus()
}
// Logout closes the connection to the server at this time and does
// nothing with the user id. In the future this will release resources
// and safely release any sensitive memory.
......@@ -770,12 +746,6 @@ type SearchCallback interface {
// Pass a callback function to extract results
func (cl *Client) SearchForUser(emailAddress string,
cb SearchCallback, timeout time.Duration) {
status := cl.commManager.GetConnectionStatus()
if status == io.Connecting || status == io.Offline {
err := errors.New("Could not SearchForUser - connection is either offline or connecting")
cb.Callback(nil, nil, err)
}
//see if the user has been searched before, if it has, return it
uid, pk := cl.session.GetContactByValue(emailAddress)
......@@ -864,11 +834,6 @@ func (cl *Client) DeleteUser(u *id.User) (string, error) {
func (cl *Client) LookupNick(user *id.User,
cb NickLookupCallback) {
go func() {
status := cl.commManager.GetConnectionStatus()
if status == io.Connecting || status == io.Offline {
err := errors.New("Could not RegisterWithUDB - connection is either offline or connecting")
cb.Callback("", err)
}
nick, err := bots.LookupNick(user)
if err != nil {
globals.Log.INFO.Printf("Lookup for nickname for user %s failed", user)
......
api/mockserver_test.go
+ 6
− 12
View file @ 1d255384
......@@ -55,8 +55,7 @@ func TestMain(m *testing.M) {
func TestRegister_ValidPrecannedRegCodeReturnsZeroID(t *testing.T) {
// Initialize client with dummy storage
storage := DummyStorage{LocationA: "Blah", StoreA: []byte{'a', 'b', 'c'}}
client, err := NewClient(&storage, "hello", "", def,
dummyConnectionStatusHandler)
client, err := NewClient(&storage, "hello", "", def)
if err != nil {
t.Errorf("Failed to initialize dummy client: %s", err.Error())
}
......@@ -88,8 +87,7 @@ func TestRegister_ValidPrecannedRegCodeReturnsZeroID(t *testing.T) {
func TestRegister_ValidRegParams___(t *testing.T) {
// Initialize client with dummy storage
storage := DummyStorage{LocationA: "Blah", StoreA: []byte{'a', 'b', 'c'}}
client, err := NewClient(&storage, "hello", "", def,
dummyConnectionStatusHandler)
client, err := NewClient(&storage, "hello", "", def)
if err != nil {
t.Errorf("Failed to initialize dummy client: %s", err.Error())
}
......@@ -125,8 +123,7 @@ func TestRegister_ValidRegParams___(t *testing.T) {
func TestRegister_InvalidPrecannedRegCodeReturnsError(t *testing.T) {
// Initialize client with dummy storage
storage := DummyStorage{LocationA: "Blah", StoreA: []byte{'a', 'b', 'c'}}
client, err := NewClient(&storage, "hello", "", def,
dummyConnectionStatusHandler)
client, err := NewClient(&storage, "hello", "", def)
if err != nil {
t.Errorf("Failed to initialize dummy client: %s", err.Error())
}
......@@ -152,8 +149,7 @@ func TestRegister_InvalidPrecannedRegCodeReturnsError(t *testing.T) {
func TestRegister_DeletedUserReturnsErr(t *testing.T) {
// Initialize client with dummy storage
storage := DummyStorage{LocationA: "Blah", StoreA: []byte{'a', 'b', 'c'}}
client, err := NewClient(&storage, "hello", "", def,
dummyConnectionStatusHandler)
client, err := NewClient(&storage, "hello", "", def)
if err != nil {
t.Errorf("Failed to initialize dummy client: %s", err.Error())
}
......@@ -185,8 +181,7 @@ func TestRegister_DeletedUserReturnsErr(t *testing.T) {
func TestSend(t *testing.T) {
// Initialize client with dummy storage
storage := DummyStorage{LocationA: "Blah", StoreA: []byte{'a', 'b', 'c'}}
client, err := NewClient(&storage, "hello", "", def,
dummyConnectionStatusHandler)
client, err := NewClient(&storage, "hello", "", def)
if err != nil {
t.Errorf("Failed to initialize dummy client: %s", err.Error())
}
......@@ -264,8 +259,7 @@ func TestSend(t *testing.T) {
func TestLogout(t *testing.T) {
// Initialize client with dummy storage
storage := DummyStorage{LocationA: "Blah", StoreA: []byte{'a', 'b', 'c'}}
client, err := NewClient(&storage, "hello", "", def,
dummyConnectionStatusHandler)
client, err := NewClient(&storage, "hello", "", def)
if err != nil {
t.Errorf("Failed to initialize dummy client: %s", err.Error())
}
......
bindings/client.go
+ 0
− 9
View file @ 1d255384
......@@ -352,15 +352,6 @@ func (cl *Client) GetSessionData() ([]byte, error) {
return cl.client.GetSessionData()
}
//Call to get the networking status of the client
// 0 - Offline
// 1 - Connecting
// 2 - Connected
func (cl *Client) GetNetworkStatus() int64 {
globals.Log.INFO.Printf("Binding call: GetNetworkStatus()")
return int64(cl.client.GetNetworkStatus())
}
//LoadEncryptedSession: Spits out the encrypted session file in text
func (cl *Client) LoadEncryptedSession() (string, error) {
globals.Log.INFO.Printf("Binding call: LoadEncryptedSession()")
......
bindings/client_test.go
+ 10
− 15
View file @ 1d255384
......@@ -107,14 +107,12 @@ func TestNewClientNil(t *testing.T) {
ndfStr, pubKey := getNDFJSONStr(def, t)
_, err := NewClient(nil, "", "", ndfStr, pubKey,
&MockConStatCallback{})
_, err := NewClient(nil, "", "", ndfStr, pubKey)
if err == nil {
t.Errorf("NewClient returned nil on invalid (nil, nil) input!")
}
_, err = NewClient(nil, "", "", "", "hello",
&MockConStatCallback{})
_, err = NewClient(nil, "", "", "", "hello")
if err == nil {
t.Errorf("NewClient returned nil on invalid (nil, 'hello') input!")
}
......@@ -125,8 +123,7 @@ func TestNewClient(t *testing.T) {
ndfStr, pubKey := getNDFJSONStr(def, t)
client, err := NewClient(&d, "hello", "", ndfStr, pubKey,
&MockConStatCallback{})
client, err := NewClient(&d, "hello", "", ndfStr, pubKey)
if err != nil {
t.Errorf("NewClient returned error: %v", err)
} else if client == nil {
......@@ -142,8 +139,7 @@ func TestRegister(t *testing.T) {
ndfStr, pubKey := getNDFJSONStr(def, t)
d := DummyStorage{LocationA: "Blah", StoreA: []byte{'a', 'b', 'c'}}
client, err := NewClient(&d, "hello", "", ndfStr, pubKey,
&MockConStatCallback{})
client, err := NewClient(&d, "hello", "", ndfStr, pubKey)
client.DisableTLS()
if err != nil {
t.Errorf("Failed to marshal group JSON: %s", err)
......@@ -178,8 +174,7 @@ func TestLoginLogout(t *testing.T) {
ndfStr, pubKey := getNDFJSONStr(def, t)
d := DummyStorage{LocationA: "Blah", StoreA: []byte{'a', 'b', 'c'}}
client, err := NewClient(&d, "hello", "", ndfStr, pubKey,
&MockConStatCallback{})
client, err := NewClient(&d, "hello", "", ndfStr, pubKey)
if err != nil {
t.Errorf("Error starting client: %+v", err)
}
......@@ -225,8 +220,7 @@ func TestListen(t *testing.T) {
ndfStr, pubKey := getNDFJSONStr(def, t)
d := DummyStorage{LocationA: "Blah", StoreA: []byte{'a', 'b', 'c'}}
client, err := NewClient(&d, "hello", "", ndfStr, pubKey,
&MockConStatCallback{})
client, err := NewClient(&d, "hello", "", ndfStr, pubKey)
client.DisableTLS()
// Connect to gateway
err = client.Connect()
......@@ -266,8 +260,7 @@ func TestStopListening(t *testing.T) {
ndfStr, pubKey := getNDFJSONStr(def, t)
d := DummyStorage{LocationA: "Blah", StoreA: []byte{'a', 'b', 'c'}}
client, err := NewClient(&d, "hello", "", ndfStr, pubKey,
&MockConStatCallback{})
client, err := NewClient(&d, "hello", "", ndfStr, pubKey)
client.DisableTLS()
// Connect to gateway
err = client.Connect()
......@@ -382,10 +375,12 @@ func testMainWrapper(m *testing.M) int {
permComms = registration.StartRegistrationServer(pAddr, pHandler, nil, nil)
// Start mock gateways used by registration and defer their shutdown (may not be needed)
//the ports used are colliding between tests in GoLand when running full suite, this is a dumb fix
bump := rand.Intn(10) * 10
for i := 0; i < NumGWs; i++ {
gw := ndf.Gateway{
Address: fmtAddress(GWsStartPort + i),
Address: fmtAddress(GWsStartPort + i + bump),
}
def.Gateways = append(def.Gateways, gw)
......
cmd/root.go
+ 2
− 8
View file @ 1d255384
......@@ -106,14 +106,8 @@ func sessionInitialization() (*id.User, string, *api.Client) {
globals.Log.DEBUG.Printf(" NDF Verified")
//If no session file is passed initialize with RAM Storage
dummyConnectionStatusHandler := func(status uint32, timeout int) {
globals.Log.INFO.Printf("Network status: %+v, %+v", status, timeout)
}
if sessionFile == "" {
client, err = api.NewClient(&globals.RamStorage{}, "", "", ndfJSON,
dummyConnectionStatusHandler)
client, err = api.NewClient(&globals.RamStorage{}, "", "", ndfJSON)
if err != nil {
globals.Log.ERROR.Printf("Could Not Initialize Ram Storage: %s\n",
err.Error())
......@@ -151,7 +145,7 @@ func sessionInitialization() (*id.User, string, *api.Client) {
}
}
//Initialize client with OS Storage
client, err = api.NewClient(nil, sessionA, sessionB, ndfJSON, dummyConnectionStatusHandler)
client, err = api.NewClient(nil, sessionA, sessionB, ndfJSON)
if err != nil {
globals.Log.ERROR.Printf("Could Not Initialize OS Storage: %s\n", err.Error())
return id.ZeroID, "", nil
......
io/commManager.go
+ 14
− 66
View file @ 1d255384
......@@ -21,7 +21,6 @@ import (
"gitlab.com/elixxir/primitives/id"
"gitlab.com/elixxir/primitives/ndf"
"sync"
"sync/atomic"
"time"
)
......@@ -49,8 +48,6 @@ type CommManager struct {
receptionGatewayIndex int
// Index in the NDF of the gateway used to send messages
transmissionGatewayIndex int
//Callback which passes the connection status when it updates
connectionStatusCallback ConnectionStatusCallback
// blockTransmissions will use a mutex to prevent multiple threads from sending
// messages at the same time.
......@@ -64,18 +61,12 @@ type CommManager struct {
sendLock sync.Mutex
connectionStatus *uint32
registrationVersion string
lock sync.RWMutex
}
func NewCommManager(ndf *ndf.NetworkDefinition,
callback ConnectionStatusCallback) *CommManager {
status := uint32(0)
func NewCommManager(ndf *ndf.NetworkDefinition) *CommManager {
cm := &CommManager{
nextId: parse.IDCounter(),
collator: NewCollator(),
......@@ -87,8 +78,6 @@ func NewCommManager(ndf *ndf.NetworkDefinition,
ndf: ndf,
receptionGatewayIndex: len(ndf.Gateways) - 1,
transmissionGatewayIndex: 0,
connectionStatusCallback: callback,
connectionStatus: &status,
}
//cm.Comms.ConnectionManager.SetMaxRetries(1)
......@@ -98,14 +87,12 @@ func NewCommManager(ndf *ndf.NetworkDefinition,
// Connects to gateways using tls filepaths to create credential information
// for connection establishment
func (cm *CommManager) ConnectToGateways() error {
func (cm *CommManager) ConnectToGateways() error { // tear out
var err error
if len(cm.ndf.Gateways) < 1 {
return errors.New("could not connect due to invalid number of nodes")
}
cm.setConnectionStatus(Connecting, 0)
// connect to all gateways
var wg sync.WaitGroup
errChan := make(chan error, len(cm.ndf.Gateways))
......@@ -113,7 +100,7 @@ func (cm *CommManager) ConnectToGateways() error {
var gwCreds []byte
cm.lock.RLock()
cm.lock.RLock() // what is the purpose of this locked block
if gateway.TlsCertificate != "" && cm.tls {
gwCreds = []byte(gateway.TlsCertificate)
}
......@@ -122,8 +109,7 @@ func (cm *CommManager) ConnectToGateways() error {
cm.lock.RUnlock()
wg.Add(1)
go func() {
go func() { // Does this still need a thread?
globals.Log.INFO.Printf("Connecting to gateway %s at %s...",
gwID.String(), gwAddr)
host, err := connect.NewHost(gwAddr, gwCreds, false)
......@@ -145,7 +131,6 @@ func (cm *CommManager) ConnectToGateways() error {
} else {
errs = err
}
}
if errs != nil {
......@@ -153,13 +138,12 @@ func (cm *CommManager) ConnectToGateways() error {
}
}
cm.setConnectionStatus(Online, 0)
return nil
}
// Connects to the permissioning server, if we know about it, to get the latest
// version from it
func (cm *CommManager) UpdateRemoteVersion() error {
func (cm *CommManager) UpdateRemoteVersion() error { // need this but make getremoteversion, handle versioning in client
permissioningHost, ok := cm.Comms.GetHost(PermissioningAddrID)
if !ok {
return errors.Errorf("Failed to find permissioning host with id %s", PermissioningAddrID)
......@@ -173,12 +157,8 @@ func (cm *CommManager) UpdateRemoteVersion() error {
return nil
}
func (cm *CommManager) GetConnectionCallback() ConnectionStatusCallback {
return cm.connectionStatusCallback
}
//GetUpdatedNDF: Connects to the permissioning server to get the updated NDF from it
func (cm *CommManager) GetUpdatedNDF(currentNDF *ndf.NetworkDefinition) (*ndf.NetworkDefinition, error) {
func (cm *CommManager) GetUpdatedNDF(currentNDF *ndf.NetworkDefinition) (*ndf.NetworkDefinition, error) { // again, uses internal ndf. stay here, return results instead
//Hash the client's ndf for comparison with registration's ndf
hash := sha256.New()
......@@ -223,7 +203,7 @@ func (cm *CommManager) GetUpdatedNDF(currentNDF *ndf.NetworkDefinition) (*ndf.Ne
// Update NDF modifies the network properties for the network which is
// communicated with
func (cm *CommManager) UpdateNDF(updatedNDF *ndf.NetworkDefinition) {
func (cm *CommManager) UpdateNDF(updatedNDF *ndf.NetworkDefinition) { // again, don't worry about ndf in this object
cm.lock.Lock()
cm.ndf = updatedNDF
cm.receptionGatewayIndex = len(cm.ndf.Gateways) - 1
......@@ -233,14 +213,14 @@ func (cm *CommManager) UpdateNDF(updatedNDF *ndf.NetworkDefinition) {
// Utility method, returns whether the local version and remote version are
// compatible
func (cm *CommManager) CheckVersion() (bool, error) {
func (cm *CommManager) CheckVersion() (bool, error) { // again, version stuff, move to globals
return checkVersion(globals.SEMVER, cm.registrationVersion)
}
// There's currently no need to keep connected to permissioning constantly,
// so we have functions to connect to and disconnect from it when a connection
// to permissioning is needed
func (cm *CommManager) ConnectToPermissioning() (connected bool, err error) {
func (cm *CommManager) ConnectToPermissioning() (connected bool, err error) { // this disappears, make host in simple call
if cm.ndf.Registration.Address != "" {
_, ok := cm.Comms.GetHost(PermissioningAddrID)
if ok {
......@@ -273,54 +253,22 @@ func (cm *CommManager) ConnectToPermissioning() (connected bool, err error) {
}
}
func (cm *CommManager) Disconnect() {
func (cm *CommManager) Disconnect() { // gone
cm.Comms.DisconnectAll()
}
func (cm *CommManager) DisableTLS() {
status := atomic.LoadUint32(cm.connectionStatus)
if status != Setup {
globals.Log.FATAL.Panicf("Cannot disable TLS" +
"while communications are running")
}
func (cm *CommManager) DisableTLS() { // gone
cm.tls = false
}
func (cm *CommManager) GetRegistrationVersion() string {
func (cm *CommManager) GetRegistrationVersion() string { // on client
return cm.registrationVersion
}
func (cm *CommManager) DisableBlockingTransmission() {
status := atomic.LoadUint32(cm.connectionStatus)
if status != Setup {
globals.Log.FATAL.Panicf("Cannot set tramsmission to blocking" +
"while communications are running")
}
func (cm *CommManager) DisableBlockingTransmission() { // flag passed into receiver
cm.blockTransmissions = false
}
func (cm *CommManager) SetRateLimit(delay time.Duration) {
status := atomic.LoadUint32(cm.connectionStatus)
if status != Setup {
globals.Log.FATAL.Panicf("Cannot set the connection rate limit " +
"while communications are running")
}
func (cm *CommManager) SetRateLimit(delay time.Duration) { // pass into received
cm.transmitDelay = delay
}
func (cm *CommManager) GetConnectionStatus() uint32 {
return atomic.LoadUint32(cm.connectionStatus)
}
func (cm *CommManager) setConnectionStatus(status uint32, timeout int) {
atomic.SwapUint32(cm.connectionStatus, status)
globals.Log.INFO.Printf("Connection status changed to: %v", status)
go cm.connectionStatusCallback(status, timeout)
}
func toSeconds(duration time.Duration) int {
return int(duration) / int(time.Second)
}
io/connectionState.go deleted 100644 → 0
+ 0
− 8
View file @ 7ab227bc
package io
const (
Setup uint32 = iota
Offline
Connecting
Online
)
io/connectionStatusCallback.go deleted 100644 → 0
+ 0
− 3
View file @ 7ab227bc
package io
type ConnectionStatusCallback func(status uint32, TimeoutSeconds int)
io/interface.go
+ 1
− 1
View file @ 1d255384
......@@ -15,7 +15,7 @@ import (
)
// Communication interface implements send/receive functionality with the server
type Communications interface {
type Communications interface { // this can go
// SendMessage to the server
// TODO(nen) Can we get rid of the crypto type param here?
......
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment