Repository: seqsense/aws-iot-device-sdk-go
Pull request #70 (2020-04-21 05:26:30)

Add shadow
- [x] after #54 
- [x] after #77

codecov-io: # [Codecov](https://codecov.io/gh/seqsense/aws-iot-device-sdk-go/pull/70?src=pr&el=h1) Report
> Merging [#70](https://codecov.io/gh/seqsense/aws-iot-device-sdk-go/pull/70?src=pr&el=desc) into [master](https://codecov.io/gh/seqsense/aws-iot-device-sdk-go/commit/e6192c96d90d6e00522cdf42bb342ce11ed7a2c1&el=desc) will **decrease** coverage by `38.87%`.
> The diff coverage is `13.45%`.

[![Impacted file tree graph](https://codecov.io/gh/seqsense/aws-iot-device-sdk-go/pull/70/graphs/tree.svg?width=650&height=150&src=pr&token=St5U6jzMqv)](https://codecov.io/gh/seqsense/aws-iot-device-sdk-go/pull/70?src=pr&el=tree)

```diff
@@             Coverage Diff             @@
##           master      #70       +/-   ##
===========================================
- Coverage   65.51%   26.63%   -38.88%     
===========================================
  Files           2        4        +2     
  Lines          58      229      +171     
===========================================
+ Hits           38       61       +23     
- Misses         15      159      +144     
- Partials        5        9        +4     
```


| [Impacted Files](https://codecov.io/gh/seqsense/aws-iot-device-sdk-go/pull/70?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [shadow/shadow.go](https://codecov.io/gh/seqsense/aws-iot-device-sdk-go/pull/70/diff?src=pr&el=tree#diff-c2hhZG93L3NoYWRvdy5nbw==) | `0.00% <0.00%> (ø)` | |
| [shadow/state.go](https://codecov.io/gh/seqsense/aws-iot-device-sdk-go/pull/70/diff?src=pr&el=tree#diff-c2hhZG93L3N0YXRlLmdv) | `46.93% <46.93%> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/seqsense/aws-iot-device-sdk-go/pull/70?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/seqsense/aws-iot-device-sdk-go/pull/70?src=pr&el=footer). Last update [e6192c9...15e05ab](https://codecov.io/gh/seqsense/aws-iot-device-sdk-go/pull/70?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

at-wat: @kt3k @t-asaka please take a look
at-wat: @kt3k PTAL

Patch:
diff --git a/shadow/shadow.go b/shadow/shadow.go
new file mode 100644
index 0000000..3eecaed
--- /dev/null
+++ b/shadow/shadow.go
@@ -0,0 +1,232 @@
+package shadow
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"sync"
+
+	"github.com/at-wat/mqtt-go"
+	"github.com/seqsense/aws-iot-device-sdk-go/v4"
+)
+
+// Shadow is an interface of Thing Shadow.
+type Shadow interface {
+	mqtt.Handler
+	// Get thing state.
+	Get(ctx context.Context) error
+	// Report thing state.
+	Report(ctx context.Context, state interface{}) error
+	// Desire sets desired thing state.
+	Desire(ctx context.Context, state interface{}) error
+	// Document returns full thing document.
+	Document() *ThingDocument
+	// Delete thing shadow.
+	Delete(ctx context.Context) error
+	// OnDelta sets handler of state deltas.
+	OnDelta(func(delta map[string]interface{}))
+	// OnError sets handler of asynchronous errors.
+	OnError(func(error))
+}
+
+type shadow struct {
+	mqtt.ServeMux
+	cli       mqtt.Client
+	thingName string
+	doc       *ThingDocument
+	mu        sync.Mutex
+	onDelta   func(delta map[string]interface{})
+	onError   func(err error)
+}
+
+func (s *shadow) topic(operation string) string {
+	return "$aws/things/" + s.thingName + "/shadow/" + operation
+}
+
+func (s *shadow) getAccepted(msg *mqtt.Message) {
+	doc := &ThingDocument{}
+	if err := json.Unmarshal(msg.Payload, doc); err != nil {
+		s.handleError(err)
+		return
+	}
+	s.doc = doc
+	s.handleDelta(s.doc.State.Delta)
+}
+
+func (s *shadow) getRejected(msg *mqtt.Message) {
+	s.handleError(fmt.Errorf("%s: %s", msg.Topic, string(msg.Payload)))
+}
+
+func (s *shadow) updateAccepted(msg *mqtt.Message) {
+	doc := &thingDocumentRaw{}
+	if err := json.Unmarshal(msg.Payload, doc); err != nil {
+		s.handleError(err)
+		return
+	}
+	if err := s.doc.update(doc); err != nil {
+		s.handleError(err)
+		return
+	}
+}
+
+func (s *shadow) updateDelta(msg *mqtt.Message) {
+	state := &thingDelta{}
+	if err := json.Unmarshal(msg.Payload, state); err != nil {
+		s.handleError(err)
+		return
+	}
+	ok := s.doc.updateDelta(state)
+	if ok {
+		s.handleDelta(s.doc.State.Delta)
+	}
+}
+
+func (s *shadow) updateRejected(msg *mqtt.Message) {
+	s.handleError(fmt.Errorf("%s: %s", msg.Topic, string(msg.Payload)))
+}
+
+func (s *shadow) deleteAccepted(msg *mqtt.Message) {
+	s.doc = nil
+}
+
+func (s *shadow) deleteRejected(msg *mqtt.Message) {
+	s.handleError(fmt.Errorf("%s: %s", msg.Topic, string(msg.Payload)))
+}
+
+// New creates Thing Shadow interface.
+func New(ctx context.Context, cli awsiotdev.Device) (Shadow, error) {
+	s := &shadow{
+		cli:       cli,
+		thingName: cli.ThingName(),
+	}
+	for _, sub := range []struct {
+		topic   string
+		handler mqtt.Handler
+	}{
+		{s.topic("update/delta"), mqtt.HandlerFunc(s.updateDelta)},
+		{s.topic("update/accepted"), mqtt.HandlerFunc(s.updateAccepted)},
+		{s.topic("update/rejected"), mqtt.HandlerFunc(s.updateRejected)},
+		{s.topic("get/accepted"), mqtt.HandlerFunc(s.getAccepted)},
+		{s.topic("get/rejected"), mqtt.HandlerFunc(s.getRejected)},
+		{s.topic("delete/accepted"), mqtt.HandlerFunc(s.deleteAccepted)},
+		{s.topic("delete/rejected"), mqtt.HandlerFunc(s.deleteRejected)},
+	} {
+		if err := s.ServeMux.Handle(sub.topic, sub.handler); err != nil {
+			return nil, err
+		}
+	}
+
+	err := cli.Subscribe(ctx,
+		mqtt.Subscription{Topic: s.topic("update/delta"), QoS: mqtt.QoS1},
+		mqtt.Subscription{Topic: s.topic("update/accepted"), QoS: mqtt.QoS1},
+		mqtt.Subscription{Topic: s.topic("update/rejected"), QoS: mqtt.QoS1},
+		mqtt.Subscription{Topic: s.topic("get/accepted"), QoS: mqtt.QoS1},
+		mqtt.Subscription{Topic: s.topic("get/rejected"), QoS: mqtt.QoS1},
+		mqtt.Subscription{Topic: s.topic("delete/accepted"), QoS: mqtt.QoS1},
+		mqtt.Subscription{Topic: s.topic("delete/rejected"), QoS: mqtt.QoS1},
+	)
+	if err != nil {
+		return nil, err
+	}
+	return s, nil
+}
+
+func (s *shadow) Report(ctx context.Context, state interface{}) error {
+	rawState, err := json.Marshal(state)
+	if err != nil {
+		return err
+	}
+	rawStateJSON := json.RawMessage(rawState)
+	data, err := json.Marshal(&thingDocumentRaw{
+		State: thingStateRaw{Reported: rawStateJSON},
+	})
+	if err != nil {
+		return err
+	}
+	if err := s.cli.Publish(ctx, &mqtt.Message{
+		Topic:   s.topic("update"),
+		QoS:     mqtt.QoS1,
+		Payload: data,
+	}); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (s *shadow) Desire(ctx context.Context, state interface{}) error {
+	rawState, err := json.Marshal(state)
+	if err != nil {
+		return err
+	}
+	rawStateJSON := json.RawMessage(rawState)
+	data, err := json.Marshal(&thingDocumentRaw{
+		State: thingStateRaw{Desired: rawStateJSON},
+	})
+	if err != nil {
+		return err
+	}
+	if err := s.cli.Publish(ctx, &mqtt.Message{
+		Topic:   s.topic("update"),
+		QoS:     mqtt.QoS1,
+		Payload: data,
+	}); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (s *shadow) Get(ctx context.Context) error {
+	if err := s.cli.Publish(ctx, &mqtt.Message{
+		Topic:   s.topic("get"),
+		QoS:     mqtt.QoS1,
+		Payload: []byte{},
+	}); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (s *shadow) Delete(ctx context.Context) error {
+	if err := s.cli.Publish(ctx, &mqtt.Message{
+		Topic:   s.topic("delete"),
+		QoS:     mqtt.QoS1,
+		Payload: []byte{},
+	}); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (s *shadow) Document() *ThingDocument {
+	return s.doc
+}
+
+func (s *shadow) OnDelta(cb func(delta map[string]interface{})) {
+	s.mu.Lock()
+	s.onDelta = cb
+	s.mu.Unlock()
+}
+
+func (s *shadow) handleDelta(delta map[string]interface{}) {
+	s.mu.Lock()
+	cb := s.onDelta
+	s.mu.Unlock()
+	if cb != nil {
+		cb(delta)
+	}
+}
+
+func (s *shadow) OnError(cb func(err error)) {
+	s.mu.Lock()
+	s.onError = cb
+	s.mu.Unlock()
+}
+
+func (s *shadow) handleError(err error) {
+	s.mu.Lock()
+	cb := s.onError
+	s.mu.Unlock()
+	if cb != nil {
+		cb(err)
+	}
+}
diff --git a/shadow/state.go b/shadow/state.go
new file mode 100644
index 0000000..6001064
--- /dev/null
+++ b/shadow/state.go
@@ -0,0 +1,117 @@
+package shadow
+
+import (
+	"encoding/json"
+	"errors"
+)
+
+// ErrVersionConflict means thing state update was aborted due to version conflict.
+var ErrVersionConflict = errors.New("version conflict")
+
+// ThingState represents Thing Shadow State.
+type ThingState struct {
+	Desired  map[string]interface{} `json:"desired,omitempty"`
+	Reported map[string]interface{} `json:"reported,omitempty"`
+	Delta    map[string]interface{} `json:"delta,omitempty"`
+}
+
+// ThingDocument represents Thing Shadow Document.
+type ThingDocument struct {
+	State     ThingState `json:"state"`
+	Version   int        `json:"version,omitempty"`
+	Timestamp int        `json:"timestamp,omitempty"`
+}
+
+type thingStateRaw struct {
+	Desired  json.RawMessage `json:"desired,omitempty"`
+	Reported json.RawMessage `json:"reported,omitempty"`
+	Delta    json.RawMessage `json:"delta,omitempty"`
+}
+
+type thingDocumentRaw struct {
+	State     thingStateRaw `json:"state"`
+	Version   int           `json:"version,omitempty"`
+	Timestamp int           `json:"timestamp,omitempty"`
+}
+
+type thingDelta struct {
+	State     map[string]interface{} `json:"state"`
+	Version   int                    `json:"version,omitempty"`
+	Timestamp int                    `json:"timestamp,omitempty"`
+}
+
+func (s *ThingDocument) update(state *thingDocumentRaw) error {
+	if s.Version > state.Version {
+		// Received an old version; just ignore it.
+		return nil
+	}
+	s.Version = state.Version
+	s.Timestamp = state.Timestamp
+	if err := updateStateRaw(s.State.Desired, state.State.Desired); err != nil {
+		return err
+	}
+	if err := updateStateRaw(s.State.Reported, state.State.Reported); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (s *ThingDocument) updateDelta(state *thingDelta) bool {
+	if s.Version > state.Version {
+		// Received an old version; just ignore it.
+		return false
+	}
+	s.Version = state.Version
+	s.Timestamp = state.Timestamp
+	s.State.Delta = state.State
+	return true
+}
+
+func updateStateRaw(state map[string]interface{}, update json.RawMessage) error {
+	if !hasUpdate(update) {
+		return nil
+	}
+	if update == nil {
+		for k := range state {
+			delete(state, k)
+		}
+		return nil
+	}
+	var u map[string]interface{}
+	if err := json.Unmarshal([]byte(update), &u); err != nil {
+		return err
+	}
+	return updateState(state, u)
+}
+
+func updateState(state map[string]interface{}, update map[string]interface{}) error {
+	if len(update) == 0 {
+		for k := range state {
+			delete(state, k)
+		}
+		return nil
+	}
+	for key, val := range update {
+		switch v := val.(type) {
+		case map[string]interface{}:
+			if s, ok := state[key].(map[string]interface{}); ok {
+				if err := updateState(s, v); err != nil {
+					return err
+				}
+			} else {
+				state[key] = v
+			}
+		case nil:
+			if _, ok := state[key]; ok {
+				delete(state, key)
+			}
+		default:
+			state[key] = v
+		}
+	}
+	return nil
+}
+
+func hasUpdate(s json.RawMessage) bool {
+	return len(s) != 0
+}


Test patch:
diff --git a/shadow/shadow_test.go b/shadow/shadow_test.go
new file mode 100644
index 0000000..ad97896
--- /dev/null
+++ b/shadow/shadow_test.go
@@ -0,0 +1,314 @@
+package shadow
+
+import (
+	"reflect"
+	"testing"
+
+	mqtt "github.com/at-wat/mqtt-go"
+)
+
+func TestShadow(t *testing.T) {
+	t.Run("Rejected", func(t *testing.T) {
+		chErr := make(chan error, 1)
+		s := &shadow{
+			doc: &ThingDocument{
+				State: ThingState{
+					Desired: map[string]interface{}{"key": "value_init"},
+				},
+				Version:   2,
+				Timestamp: 12345,
+			},
+			onDelta: func(delta map[string]interface{}) {
+				t.Error("onDelta must not be called on rejected")
+			},
+			onError: func(err error) {
+				chErr <- err
+			},
+		}
+		expectedDoc := &ThingDocument{
+			State: ThingState{
+				Desired: map[string]interface{}{"key": "value_init"},
+			},
+			Version:   2,
+			Timestamp: 12345,
+		}
+
+		t.Run("Get", func(t *testing.T) {
+			s.getRejected(&mqtt.Message{})
+			select {
+			case err := <-chErr:
+				if err == nil {
+					t.Error("onError must be called with non-nil error")
+				}
+			default:
+				t.Fatal("Timeout")
+			}
+			if !reflect.DeepEqual(*expectedDoc, *s.doc) {
+				t.Error("Document must not be changed on reject")
+			}
+		})
+		t.Run("Update", func(t *testing.T) {
+			s.updateRejected(&mqtt.Message{})
+			select {
+			case err := <-chErr:
+				if err == nil {
+					t.Error("onError must be called with non-nil error")
+				}
+			default:
+				t.Fatal("Timeout")
+			}
+			if !reflect.DeepEqual(*expectedDoc, *s.doc) {
+				t.Error("Document must not be changed on reject")
+			}
+		})
+		t.Run("Delete", func(t *testing.T) {
+			s.deleteRejected(&mqtt.Message{})
+			select {
+			case err := <-chErr:
+				if err == nil {
+					t.Error("onError must be called with non-nil error")
+				}
+			default:
+				t.Fatal("Timeout")
+			}
+			if !reflect.DeepEqual(*expectedDoc, *s.doc) {
+				t.Error("Document must not be changed on reject")
+			}
+		})
+	})
+
+	t.Run("Accepted", func(t *testing.T) {
+		t.Run("Get", func(t *testing.T) {
+			chErr := make(chan error, 1)
+			chDelta := make(chan map[string]interface{}, 1)
+			s := &shadow{
+				doc: &ThingDocument{
+					State: ThingState{
+						Desired: map[string]interface{}{"key": "value_init"},
+					},
+					Version:   2,
+					Timestamp: 12345,
+				},
+				onDelta: func(delta map[string]interface{}) {
+					chDelta <- delta
+				},
+				onError: func(err error) {
+					chErr <- err
+				},
+			}
+			expectedDoc := &ThingDocument{
+				State: ThingState{
+					Desired: map[string]interface{}{"key2": "value2"},
+					Delta:   map[string]interface{}{"key2": "value2"},
+				},
+				Version:   3,
+				Timestamp: 12346,
+			}
+			s.getAccepted(&mqtt.Message{Payload: []byte(
+				"{" +
+					"  \"state\": {" +
+					"    \"desired\": {\"key2\": \"value2\"}," +
+					"    \"delta\": {\"key2\": \"value2\"}" +
+					"  }," +
+					"  \"version\": 3," +
+					"  \"timestamp\": 12346" +
+					"}",
+			)})
+
+			select {
+			case err := <-chErr:
+				t.Error(err)
+			case delta := <-chDelta:
+				if !reflect.DeepEqual(expectedDoc.State.Delta, delta) {
+					t.Errorf("Expected delta: %v, got: %v",
+						expectedDoc.State.Delta, delta,
+					)
+				}
+			default:
+				t.Fatal("Timeout")
+			}
+			if !reflect.DeepEqual(expectedDoc, s.doc) {
+				t.Errorf("Expected state: %v, got: %v",
+					expectedDoc, s.doc,
+				)
+			}
+		})
+		t.Run("Delete", func(t *testing.T) {
+			chErr := make(chan error, 1)
+			s := &shadow{
+				doc: &ThingDocument{
+					State: ThingState{
+						Desired: map[string]interface{}{"key": "value_init"},
+					},
+					Version:   2,
+					Timestamp: 12345,
+				},
+				onDelta: func(delta map[string]interface{}) {
+					t.Error("Delete must not trigger onDelta")
+				},
+				onError: func(err error) {
+					chErr <- err
+				},
+			}
+			s.deleteAccepted(&mqtt.Message{})
+
+			select {
+			case err := <-chErr:
+				t.Error(err)
+			default:
+			}
+			if s.doc != nil {
+				t.Errorf("Document must be nil after delete")
+			}
+		})
+	})
+
+	t.Run("Update", func(t *testing.T) {
+		t.Run("Delta", func(t *testing.T) {
+			chErr := make(chan error, 1)
+			chDelta := make(chan map[string]interface{}, 1)
+			s := &shadow{
+				doc: &ThingDocument{
+					State: ThingState{
+						Desired: map[string]interface{}{"key2": "value2"},
+					},
+					Version:   2,
+					Timestamp: 12345,
+				},
+				onDelta: func(delta map[string]interface{}) {
+					chDelta <- delta
+				},
+				onError: func(err error) {
+					chErr <- err
+				},
+			}
+			expectedDoc := &ThingDocument{
+				State: ThingState{
+					Desired: map[string]interface{}{"key2": "value2"},
+					Delta:   map[string]interface{}{"key2": "value2"},
+				},
+				Version:   3,
+				Timestamp: 12346,
+			}
+			s.updateDelta(&mqtt.Message{Payload: []byte(
+				"{" +
+					"  \"state\": {" +
+					"    \"key2\": \"value2\"" +
+					"  }," +
+					"  \"version\": 3," +
+					"  \"timestamp\": 12346" +
+					"}",
+			)})
+
+			select {
+			case err := <-chErr:
+				t.Error(err)
+			case delta := <-chDelta:
+				if !reflect.DeepEqual(expectedDoc.State.Delta, delta) {
+					t.Errorf("Expected delta: %v, got: %v",
+						expectedDoc.State.Delta, delta,
+					)
+				}
+			default:
+				t.Fatal("Timeout")
+			}
+			if !reflect.DeepEqual(expectedDoc, s.doc) {
+				t.Errorf("Expected state: %v, got: %v",
+					expectedDoc, s.doc,
+				)
+			}
+		})
+		t.Run("Accepted", func(t *testing.T) {
+			chErr := make(chan error, 1)
+			chDelta := make(chan map[string]interface{}, 1)
+			s := &shadow{
+				doc: &ThingDocument{
+					State: ThingState{
+						Reported: map[string]interface{}{"key1": "value1"},
+					},
+					Version:   2,
+					Timestamp: 12345,
+				},
+				onDelta: func(delta map[string]interface{}) {
+					chDelta <- delta
+				},
+				onError: func(err error) {
+					chErr <- err
+				},
+			}
+			expectedDoc := &ThingDocument{
+				State: ThingState{
+					Reported: map[string]interface{}{
+						"key1": "value1",
+						"key2": "value2",
+					},
+				},
+				Version:   3,
+				Timestamp: 12346,
+			}
+			s.updateAccepted(&mqtt.Message{Payload: []byte(
+				"{" +
+					"  \"state\": {" +
+					"    \"Reported\": {" +
+					"      \"key2\": \"value2\"" +
+					"    }" +
+					"  }," +
+					"  \"version\": 3," +
+					"  \"timestamp\": 12346" +
+					"}",
+			)})
+
+			select {
+			case err := <-chErr:
+				t.Error(err)
+			case <-chDelta:
+				t.Error("cbDelta must not be called on update accept")
+			default:
+			}
+			if !reflect.DeepEqual(expectedDoc, s.doc) {
+				t.Errorf("Expected state: %v, got: %v",
+					expectedDoc, s.doc,
+				)
+			}
+		})
+		t.Run("OldDelta", func(t *testing.T) {
+			s := &shadow{
+				doc: &ThingDocument{
+					State: ThingState{
+						Desired: map[string]interface{}{"key2": "value2"},
+					},
+					Version:   2,
+					Timestamp: 12345,
+				},
+				onDelta: func(delta map[string]interface{}) {
+					t.Error("Old delta must be discarded")
+				},
+				onError: func(err error) {
+					t.Error("Old delta must be silently discarded")
+				},
+			}
+			expectedDoc := &ThingDocument{
+				State: ThingState{
+					Desired: map[string]interface{}{"key2": "value2"},
+				},
+				Version:   2,
+				Timestamp: 12345,
+			}
+			s.updateDelta(&mqtt.Message{Payload: []byte(
+				"{" +
+					"  \"state\": {" +
+					"    \"key2\": \"value\"" +
+					"  }," +
+					"  \"version\": 1," +
+					"  \"timestamp\": 12343" +
+					"}",
+			)})
+
+			if !reflect.DeepEqual(expectedDoc, s.doc) {
+				t.Errorf("Expected state: %v, got: %v",
+					expectedDoc, s.doc,
+				)
+			}
+		})
+	})
+}
diff --git a/shadow/state_test.go b/shadow/state_test.go
new file mode 100644
index 0000000..722efea
--- /dev/null
+++ b/shadow/state_test.go
@@ -0,0 +1,75 @@
+package shadow
+
+import (
+	"encoding/json"
+	"reflect"
+	"testing"
+)
+
+func TestThingDocument_update(t *testing.T) {
+	doc := &ThingDocument{
+		State:     ThingState{Desired: map[string]interface{}{"key": "value_init"}},
+		Version:   2,
+		Timestamp: 12345,
+	}
+
+	if err := doc.update(&thingDocumentRaw{
+		State:     thingStateRaw{Desired: json.RawMessage(`{"key": "ignored"}`)},
+		Version:   1,
+		Timestamp: 0,
+	}); err != nil {
+		t.Fatal(err)
+	}
+
+	expected0 := &ThingDocument{
+		State:     ThingState{Desired: map[string]interface{}{"key": "value_init"}},
+		Version:   2,
+		Timestamp: 12345,
+	}
+	if !reflect.DeepEqual(*expected0, *doc) {
+		t.Errorf(
+			"Old version must be discarded\nexpected: %v\ngot: %v",
+			*expected0, *doc,
+		)
+	}
+
+	if err := doc.update(&thingDocumentRaw{
+		State:     thingStateRaw{Desired: json.RawMessage(`{"key": "new value"}`)},
+		Version:   4,
+		Timestamp: 12346,
+	}); err != nil {
+		t.Fatal(err)
+	}
+
+	expected1 := &ThingDocument{
+		State:     ThingState{Desired: map[string]interface{}{"key": "new value"}},
+		Version:   4,
+		Timestamp: 12346,
+	}
+	if !reflect.DeepEqual(*expected1, *doc) {
+		t.Errorf(
+			"Document must be update by new version\nexpected: %v\ngot: %v",
+			*expected1, *doc,
+		)
+	}
+
+	if err := doc.update(&thingDocumentRaw{
+		State:     thingStateRaw{Desired: json.RawMessage(`null`)},
+		Version:   5,
+		Timestamp: 12347,
+	}); err != nil {
+		t.Fatal(err)
+	}
+
+	expected2 := &ThingDocument{
+		State:     ThingState{Desired: map[string]interface{}{}},
+		Version:   5,
+		Timestamp: 12347,
+	}
+	if !reflect.DeepEqual(*expected2, *doc) {
+		t.Errorf(
+			"Document must be cleared by nil\nexpected: %v\ngot: %v",
+			*expected2, *doc,
+		)
+	}
+}