fixed bug where duplicated device would remove itself and the newly added device

device/manager.go

@@ -250,8 +250,11 @@ func (m *manager) dispatch(e *Event) {
 // dispatches message failed events for any messages that were waiting to be delivered
 // at the time of pump closure.
 func (m *manager) pumpClose(d *device, c io.Closer, reason CloseReason) {
-	// remove will invoke requestClose()
-	m.devices.remove(d.id, reason)
+
+	if !m.isDeviceDuplicated(d) {
+		// remove will invoke requestClose()
+		m.devices.remove(d.id, reason)
+	}
 
 	closeError := c.Close()
 
@@ -490,3 +493,11 @@ func (m *manager) Route(request *Request) (*Response, error) {
 		return nil, ErrorDeviceNotFound
 	}
 }
+
+func (m *manager) isDeviceDuplicated(d *device) bool {
+	existing, ok := m.devices.get(d.id)
+	if !ok {
+		return false
+	}
+	return existing.state != d.state
+}

device/manager_test.go

@@ -395,3 +395,50 @@ func TestGaugeCardinality(t *testing.T) {
 		m.(*manager).measures.Models.With("neat", "bad").Add(-1)
 	})
 }
+
+func TestManagerIsDeviceDuplicated(t *testing.T) {
+	var(
+		assert = assert.New(t)
+		tests  = []struct {
+			expected	bool
+			existing	*device
+			new			*device
+			m			*manager
+		} {
+			{
+				expected: false,
+				existing: nil,
+				new:	  &device{id:"test"},
+				m: 		  NewManager(&Options{
+					MaxDevices: 0,
+				}).(*manager),
+			},
+			{
+				expected: false,
+				existing: &device{id:"test", state:stateOpen},
+				new:	  &device{id:"test", state:stateOpen},
+				m: 		  NewManager(&Options{
+					MaxDevices: 0,
+				}).(*manager),
+			},
+			{
+				expected: true,
+				existing: &device{id:"test", state:stateOpen},
+				new:	  &device{id:"test", state:stateClosed},
+				m: 		  NewManager(&Options{
+					MaxDevices: 0,
+				}).(*manager),
+			},
+		}
+	)
+
+	for _, test := range tests {
+		if test.existing != nil {
+			err := test.m.devices.add(test.existing)
+			if err != nil {
+				assert.Error(err)
+			}
+		}
+		assert.Equal(test.expected, test.m.isDeviceDuplicated(test.new))
+	}
+}