Add cancellation to watchable values

go.mod

@@ -24,7 +24,7 @@ require (
 	github.com/srwiley/oksvg v0.0.0-20210519022825-9fc0c575d5fe // indirect
 	github.com/srwiley/rasterx v0.0.0-20210519020934-456a8d69b780 // indirect
 	github.com/yuin/goldmark v1.4.1 // indirect
-	go.arsenm.dev/infinitime v0.0.0-20211022195951-45baea10486b
+	go.arsenm.dev/infinitime v0.0.0-20211023042633-53aa6f8a0c72
 	golang.org/x/image v0.0.0-20210628002857-a66eb6448b8d // indirect
 	golang.org/x/net v0.0.0-20211011170408-caeb26a5c8c0 // indirect
 	golang.org/x/sys v0.0.0-20211007075335-d3039528d8ac // indirect

go.sum

@@ -366,8 +366,8 @@ github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1
 github.com/yuin/goldmark v1.3.8/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
 github.com/yuin/goldmark v1.4.1 h1:/vn0k+RBvwlxEmP5E7SZMqNxPhfMVFEJiykr15/0XKM=
 github.com/yuin/goldmark v1.4.1/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
-go.arsenm.dev/infinitime v0.0.0-20211022195951-45baea10486b h1:2VitKPwSYSWXmL5BH88nfTPLSIYPCt4yubpEJHhcQBc=
+go.arsenm.dev/infinitime v0.0.0-20211023042633-53aa6f8a0c72 h1:e8kOuL6Jj8ZjJzkGwJ3xqpGG9EhUzfvZk9AlSsm3X1U=
-go.arsenm.dev/infinitime v0.0.0-20211022195951-45baea10486b/go.mod h1:gaepaueUz4J5FfxuV19B4w5pi+V3mD0LTef50ryxr/Q=
+go.arsenm.dev/infinitime v0.0.0-20211023042633-53aa6f8a0c72/go.mod h1:gaepaueUz4J5FfxuV19B4w5pi+V3mD0LTef50ryxr/Q=
 go.etcd.io/etcd/api/v3 v3.5.0/go.mod h1:cbVKeC6lCfl7j/8jBhAK6aIYO9XOjdptoxU/nLQcPvs=
 go.etcd.io/etcd/client/pkg/v3 v3.5.0/go.mod h1:IJHfcCEKxYu1Os13ZdwCwIUTUVGYTSAM3YSwc9/Ac1g=
 go.etcd.io/etcd/client/v2 v2.305.0/go.mod h1:h9puh54ZTgAKtEbut2oe9P4L/oqKCVB6xsXlzd7alYQ=

internal/types/types.go

@@ -9,11 +9,15 @@ const (
 	ReqTypeNotify
 	ReqTypeSetTime
 	ReqTypeWatchHeartRate
+	ReqTypeCancelHeartRate
 	ReqTypeWatchBattLevel
+	ReqTypeCancelBattLevel
 	ReqTypeMotion
 	ReqTypeWatchMotion
+	ReqTypeCancelMotion
 	ReqTypeStepCount
 	ReqTypeWatchStepCount
+	ReqTypeCancelStepCount
 )
 
 const (
@@ -25,11 +29,15 @@ const (
 	ResTypeNotify
 	ResTypeSetTime
 	ResTypeWatchHeartRate
+	ResTypeCancelHeartRate
 	ResTypeWatchBattLevel
+	ResTypeCancelBattLevel
 	ResTypeMotion
 	ResTypeWatchMotion
+	ResTypeCancelMotion
 	ResTypeStepCount
 	ResTypeWatchStepCount
+	ResTypeCancelStepCount
 )
 
 const (
@@ -43,7 +51,7 @@ type ReqDataFwUpgrade struct {
 }
 
 type Response struct {
-	Type    int
+	Type    int         `json:"type"`
 	Value   interface{} `json:"value,omitempty"`
 	Message string      `json:"msg,omitempty"`
 	Error   bool        `json:"error"`

socket.go

@@ -81,6 +81,11 @@ func handleConnection(conn net.Conn, dev *infinitime.Device) {
 		return
 	}
 
+	heartRateDone := make(chan struct{})
+	battLevelDone := make(chan struct{})
+	stepCountDone := make(chan struct{})
+	motionDone := make(chan struct{})
+
 	// Create new scanner on connection
 	scanner := bufio.NewScanner(conn)
 	for scanner.Scan() {
@@ -106,19 +111,31 @@ func handleConnection(conn net.Conn, dev *infinitime.Device) {
 				Value: heartRate,
 			})
 		case types.ReqTypeWatchHeartRate:
-			heartRateCh, err := dev.WatchHeartRate()
+			heartRateCh, cancel, err := dev.WatchHeartRate()
 			if err != nil {
 				connErr(conn, err, "Error getting heart rate channel")
 				break
 			}
 			go func() {
+				// For every heart rate value
 				for heartRate := range heartRateCh {
+					select {
+					case <-heartRateDone:
+						// Stop notifications if done signal received
+						cancel()
+						return
+					default:
+						// Encode response to connection if no done signal received
 						json.NewEncoder(conn).Encode(types.Response{
 							Type:  types.ResTypeWatchHeartRate,
 							Value: heartRate,
 						})
 					}
+				}
 			}()
+		case types.ReqTypeCancelHeartRate:
+			// Stop heart rate notifications
+			heartRateDone <- struct{}{}
 		case types.ReqTypeBattLevel:
 			// Get battery level from watch
 			battLevel, err := dev.BatteryLevel()
@@ -132,19 +149,31 @@ func handleConnection(conn net.Conn, dev *infinitime.Device) {
 				Value: battLevel,
 			})
 		case types.ReqTypeWatchBattLevel:
-			battLevelCh, err := dev.WatchBatteryLevel()
+			battLevelCh, cancel, err := dev.WatchBatteryLevel()
 			if err != nil {
 				connErr(conn, err, "Error getting battery level channel")
 				break
 			}
 			go func() {
+				// For every battery level value
 				for battLevel := range battLevelCh {
+					select {
+					case <-battLevelDone:
+						// Stop notifications if done signal received
+						cancel()
+						return
+					default:
+						// Encode response to connection if no done signal received
 						json.NewEncoder(conn).Encode(types.Response{
 							Type:  types.ResTypeWatchBattLevel,
 							Value: battLevel,
 						})
 					}
+				}
 			}()
+		case types.ReqTypeCancelBattLevel:
+			// Stop battery level notifications
+			battLevelDone <- struct{}{}
 		case types.ReqTypeMotion:
 			// Get battery level from watch
 			motionVals, err := dev.Motion()
@@ -158,19 +187,31 @@ func handleConnection(conn net.Conn, dev *infinitime.Device) {
 				Value: motionVals,
 			})
 		case types.ReqTypeWatchMotion:
-			motionValCh, _, err := dev.WatchMotion()
+			motionValCh, cancel, err := dev.WatchMotion()
 			if err != nil {
 				connErr(conn, err, "Error getting heart rate channel")
 				break
 			}
 			go func() {
+				// For every motion event
 				for motionVals := range motionValCh {
+					select {
+					case <-motionDone:
+						// Stop notifications if done signal received
+						cancel()
+						return
+					default:
+						// Encode response to connection if no done signal received
 						json.NewEncoder(conn).Encode(types.Response{
 							Type:  types.ResTypeWatchMotion,
 							Value: motionVals,
 						})
 					}
+				}
 			}()
+		case types.ReqTypeCancelMotion:
+			// Stop motion notifications
+			motionDone <- struct{}{}
 		case types.ReqTypeStepCount:
 			// Get battery level from watch
 			stepCount, err := dev.StepCount()
@@ -184,19 +225,31 @@ func handleConnection(conn net.Conn, dev *infinitime.Device) {
 				Value: stepCount,
 			})
 		case types.ReqTypeWatchStepCount:
-			stepCountCh, _, err := dev.WatchStepCount()
+			stepCountCh, cancel, err := dev.WatchStepCount()
 			if err != nil {
 				connErr(conn, err, "Error getting heart rate channel")
 				break
 			}
 			go func() {
+				// For every step count value
 				for stepCount := range stepCountCh {
+					select {
+					case <-stepCountDone:
+						// Stop notifications if done signal received
+						cancel()
+						return
+					default:
+						// Encode response to connection if no done signal received
 						json.NewEncoder(conn).Encode(types.Response{
 							Type:  types.ResTypeWatchStepCount,
 							Value: stepCount,
 						})
 					}
+				}
 			}()
+		case types.ReqTypeCancelStepCount:
+			// Stop step count notifications
+			stepCountDone <- struct{}{}
 		case types.ReqTypeFwVersion:
 			// Get firmware version from watch
 			version, err := dev.Version()