Feature/destroy session

cmd/di.go

@@ -284,12 +284,12 @@ func newSessionManagerFactory(
 	sessionStorage *session.StorageMemory,
 	promiseHandler func(dialog communication.Dialog) session.PromiseProcessor,
 ) session.ManagerFactory {
-	return func(dialog communication.Dialog) session.Manager {
+	return func(dialog communication.Dialog) *session.Manager {
 		return session.NewManager(
 			proposal,
 			session.GenerateUUID,
 			configProvider,
-			sessionStorage.Add,
+			sessionStorage,
 			promiseHandler(dialog),
 		)
 	}

communication/nats/receiver.go

@@ -76,7 +76,6 @@ func (receiver *receiverNATS) Receive(consumer communication.MessageConsumer) er
 }
 
 func (receiver *receiverNATS) Respond(consumer communication.RequestConsumer) error {
-
 	requestTopic := receiver.messageTopic + string(consumer.GetRequestEndpoint())
 
 	messageHandler := func(msg *nats.Msg) {
@@ -112,7 +111,13 @@ func (receiver *receiverNATS) Respond(consumer communication.RequestConsumer) er
 		}
 	}
 
+	log.Debug(receiverLogPrefix, fmt.Sprintf("Request '%s' topic has been subscribed to", requestTopic))
+
 	_, err := receiver.connection.Subscribe(requestTopic, messageHandler)
+	// TODO: nats.Subscription.Unsubscribe() should be called when topic is no longer needed
+	//  session-create and session-destroy topic should be cleared after session-destroy message is received
+	//    or session timeouts (promise processor detects session timeout)
+	// scope of "session-create topic deduplication #533"
 	if err != nil {
 		err = fmt.Errorf("failed subscribe request '%s'. %s", requestTopic, err)
 		return err

core/connection/manager.go

@@ -154,6 +154,8 @@ func (manager *connectionManager) startConnection(consumerID identity.Identity,
 		return err
 	}
 
+	cancel = append(cancel, func() { session.RequestSessionDestroy(dialog, sessionID) })
+
 	// set the session info for future use
 	manager.sessionInfo = SessionInfo{
 		SessionID:  sessionID,
@@ -196,7 +198,7 @@ func (manager *connectionManager) startConnection(consumerID identity.Identity,
 
 	if !params.DisableKillSwitch {
 		// TODO: Implement fw based kill switch for respective OS
-		// we may need to wait for tun device to bet setup
+		// we may need to wait for tun device setup to be finished
 		firewall.NewKillSwitch().Enable()
 	}
 

core/connection/stubs_test.go

@@ -18,6 +18,7 @@
 package connection
 
 import (
+	"errors"
 	"sync"
 
 	"github.com/mysteriumnetwork/node/communication"
@@ -253,13 +254,25 @@ func (fd *fakeDialog) Send(producer communication.MessageProducer) error {
 	return nil
 }
 
+var ErrUnknownRequest = errors.New("unknown request")
+
 func (fd *fakeDialog) Request(producer communication.RequestProducer) (responsePtr interface{}, err error) {
-	return &session.CreateResponse{
-			Success: true,
-			Session: session.SessionDto{
-				ID:     fd.sessionID,
-				Config: []byte("{}"),
+	if producer.GetRequestEndpoint() == communication.RequestEndpoint("session-destroy") {
+		return &session.DestroyResponse{
+				Success: true,
 			},
-		},
-		nil
+			nil
+	}
+
+	if producer.GetRequestEndpoint() == communication.RequestEndpoint("session-create") {
+		return &session.CreateResponse{
+				Success: true,
+				Session: session.SessionDto{
+					ID:     fd.sessionID,
+					Config: []byte("{}"),
+				},
+			},
+			nil
+	}
+	return nil, ErrUnknownRequest
 }

session/create_consumer.go

@@ -24,15 +24,15 @@ import (
 	"github.com/mysteriumnetwork/node/identity"
 )
 
-// Manager defines methods for session management
-type Manager interface {
-	Create(consumerID identity.Identity, proposalID int) (Session, error)
-}
-
 // createConsumer processes session create requests from communication channel.
 type createConsumer struct {
-	SessionManager Manager
-	PeerID         identity.Identity
+	sessionCreator Creator
+	peerID         identity.Identity
+}
+
+// Creator defines method for session creation
+type Creator interface {
+	Create(consumerID identity.Identity, proposalID int) (Session, error)
 }
 
 // GetMessageEndpoint returns endpoint there to receive messages
@@ -50,7 +50,7 @@ func (consumer *createConsumer) NewRequest() (requestPtr interface{}) {
 func (consumer *createConsumer) Consume(requestPtr interface{}) (response interface{}, err error) {
 	request := requestPtr.(*CreateRequest)
 
-	sessionInstance, err := consumer.SessionManager.Create(consumer.PeerID, request.ProposalId)
+	sessionInstance, err := consumer.sessionCreator.Create(consumer.peerID, request.ProposalId)
 	switch err {
 	case nil:
 		return responseWithSession(sessionInstance), nil

session/create_consumer_test.go

@@ -34,8 +34,8 @@ func TestConsumer_Success(t *testing.T) {
 		},
 	}
 	consumer := createConsumer{
-		SessionManager: mockManager,
-		PeerID:         identity.FromAddress("peer-id"),
+		sessionCreator: mockManager,
+		peerID:         identity.FromAddress("peer-id"),
 	}
 
 	request := consumer.NewRequest().(*CreateRequest)
@@ -62,7 +62,7 @@ func TestConsumer_ErrorInvalidProposal(t *testing.T) {
 	mockManager := &managerFake{
 		returnError: ErrorInvalidProposal,
 	}
-	consumer := createConsumer{SessionManager: mockManager}
+	consumer := createConsumer{sessionCreator: mockManager}
 
 	request := consumer.NewRequest().(*CreateRequest)
 	sessionResponse, err := consumer.Consume(request)
@@ -75,7 +75,7 @@ func TestConsumer_ErrorFatal(t *testing.T) {
 	mockManager := &managerFake{
 		returnError: errors.New("fatality"),
 	}
-	consumer := createConsumer{SessionManager: mockManager}
+	consumer := createConsumer{sessionCreator: mockManager}
 
 	request := consumer.NewRequest().(*CreateRequest)
 	sessionResponse, err := consumer.Consume(request)
@@ -84,7 +84,7 @@ func TestConsumer_ErrorFatal(t *testing.T) {
 	assert.Exactly(t, responseInternalError, sessionResponse)
 }
 
-// managerFake represents fake manager usually useful in tests
+// managerFake represents fake Manager usually useful in tests
 type managerFake struct {
 	lastConsumerID identity.Identity
 	lastProposalID int
@@ -98,3 +98,8 @@ func (manager *managerFake) Create(consumerID identity.Identity, proposalID int)
 	manager.lastProposalID = proposalID
 	return manager.returnSession, manager.returnError
 }
+
+// Destroy fake destroy function
+func (manager *managerFake) Destroy(consumerID identity.Identity, sessionID string) error {
+	return nil
+}

session/create_dto.go

@@ -42,7 +42,7 @@ type CreateResponse struct {
 	Session SessionDto `json:"session"`
 }
 
-// SessionDto structure represents session information data within session creation response (session id and configuration options for underlaying service type)
+// SessionDto structure represents session information data within session creation response (session id and configuration options for underlying service type)
 type SessionDto struct {
 	ID     ID              `json:"id"`
 	Config json.RawMessage `json:"config"`

session/destroy_consumer.go

@@ -0,0 +1,58 @@
+/*
+ * Copyright (C) 2018 The "MysteriumNetwork/node" Authors.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package session
+
+import (
+	"github.com/mysteriumnetwork/node/communication"
+	"github.com/mysteriumnetwork/node/identity"
+)
+
+// destroyConsumer processes session create requests from communication channel.
+type destroyConsumer struct {
+	SessionDestroyer Destroyer
+	PeerID           identity.Identity
+}
+
+// Destroyer interface for destroying session
+type Destroyer interface {
+	Destroy(consumerID identity.Identity, sessionID string) error
+}
+
+// GetMessageEndpoint returns endpoint where to receive messages
+func (consumer *destroyConsumer) GetRequestEndpoint() communication.RequestEndpoint {
+	return endpointSessionDestroy
+}
+
+// NewRequest creates struct where request from endpoint will be serialized
+func (consumer *destroyConsumer) NewRequest() (requestPtr interface{}) {
+	return &DestroyRequest{}
+}
+
+// Consume handles requests from endpoint and replies with response
+func (consumer *destroyConsumer) Consume(requestPtr interface{}) (response interface{}, err error) {
+	request := requestPtr.(*DestroyRequest)
+
+	err = consumer.SessionDestroyer.Destroy(consumer.PeerID, request.SessionID)
+	return destroyResponse(), err
+}
+
+func destroyResponse() DestroyResponse {
+	return DestroyResponse{
+		Success: true,
+	}
+}

session/destroy_consumer_test.go

@@ -0,0 +1,88 @@
+/*
+ * Copyright (C) 2018 The "MysteriumNetwork/node" Authors.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package session
+
+import (
+	"testing"
+
+	"github.com/mysteriumnetwork/node/identity"
+	"github.com/stretchr/testify/assert"
+)
+
+// managerDestroyFake represents fake destroy Manager usually useful in tests
+type managerDestroyFake struct {
+	returnSession Session
+	returnError   error
+}
+
+func TestDestroyConsumer_Success(t *testing.T) {
+	mockDestroyer := &managerDestroyFake{
+		returnSession: Session{
+			"some-session-id",
+			fakeSessionConfig{"string-param", 123},
+			identity.FromAddress("123"),
+		},
+	}
+	consumer := destroyConsumer{
+		SessionDestroyer: mockDestroyer,
+		PeerID:           identity.FromAddress("peer-id"),
+	}
+
+	request := consumer.NewRequest().(*DestroyRequest)
+	sessionResponse, err := consumer.Consume(request)
+
+	assert.NoError(t, err)
+	assert.Exactly(
+		t,
+		DestroyResponse{
+			Success: true,
+		},
+		sessionResponse,
+	)
+}
+
+func TestDestroyConsumer_ErrorInvalidSession(t *testing.T) {
+	mockDestroyer := &managerDestroyFake{
+		returnError: ErrorSessionNotExists,
+	}
+	consumer := destroyConsumer{SessionDestroyer: mockDestroyer}
+
+	request := consumer.NewRequest().(*DestroyRequest)
+	sessionResponse, err := consumer.Consume(request)
+
+	assert.Error(t, err)
+	assert.Exactly(t, destroyResponse(), sessionResponse)
+}
+
+func TestDestroyConsumer_ErrorNotSessionOwner(t *testing.T) {
+	mockDestroyer := &managerDestroyFake{
+		returnError: ErrorWrongSessionOwner,
+	}
+	consumer := destroyConsumer{SessionDestroyer: mockDestroyer}
+
+	request := consumer.NewRequest().(*DestroyRequest)
+	sessionResponse, err := consumer.Consume(request)
+
+	assert.Error(t, err)
+	assert.Exactly(t, destroyResponse(), sessionResponse)
+}
+
+// Destroy fake destroy function
+func (manager *managerDestroyFake) Destroy(consumerID identity.Identity, sessionID string) error {
+	return manager.returnError
+}

session/destroy_dto.go

@@ -0,0 +1,35 @@
+/*
+ * Copyright (C) 2018 The "MysteriumNetwork/node" Authors.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package session
+
+import (
+	"github.com/mysteriumnetwork/node/communication"
+)
+
+const endpointSessionDestroy = communication.RequestEndpoint("session-destroy")
+
+// DestroyRequest structure represents message from service consumer to destroy session for given session id
+type DestroyRequest struct {
+	SessionID string `json:"session_id"`
+}
+
+// DestroyResponse structure represents service provider response to given session request from consumer
+type DestroyResponse struct {
+	Success bool   `json:"success"`
+	Message string `json:"message"`
+}