Repository: seqsense/aws-iot-device-sdk-go
Pull request #167 (2020-10-31 13:40:35)

Wrap errors

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

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

```diff
@@            Coverage Diff             @@
##           master     #167      +/-   ##
==========================================
- Coverage   46.73%   46.69%   -0.04%     
==========================================
  Files          25       26       +1     
  Lines        1209     1227      +18     
==========================================
+ Hits          565      573       +8     
- Misses        565      574       +9     
- Partials       79       80       +1     
```


| [Impacted Files](https://codecov.io/gh/seqsense/aws-iot-device-sdk-go/pull/167?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [awsiotdev.go](https://codecov.io/gh/seqsense/aws-iot-device-sdk-go/pull/167/diff?src=pr&el=tree#diff-YXdzaW90ZGV2Lmdv) | `75.00% <0.00%> (ø)` | |
| [dialer.go](https://codecov.io/gh/seqsense/aws-iot-device-sdk-go/pull/167/diff?src=pr&el=tree#diff-ZGlhbGVyLmdv) | `59.09% <0.00%> (ø)` | |
| [jobs/jobs.go](https://codecov.io/gh/seqsense/aws-iot-device-sdk-go/pull/167/diff?src=pr&el=tree#diff-am9icy9qb2JzLmdv) | `70.21% <0.00%> (ø)` | |
| [presigner/presign.go](https://codecov.io/gh/seqsense/aws-iot-device-sdk-go/pull/167/diff?src=pr&el=tree#diff-cHJlc2lnbmVyL3ByZXNpZ24uZ28=) | `70.00% <0.00%> (ø)` | |
| [shadow/state.go](https://codecov.io/gh/seqsense/aws-iot-device-sdk-go/pull/167/diff?src=pr&el=tree#diff-c2hhZG93L3N0YXRlLmdv) | `57.14% <0.00%> (ø)` | |
| [tunnel/destination.go](https://codecov.io/gh/seqsense/aws-iot-device-sdk-go/pull/167/diff?src=pr&el=tree#diff-dHVubmVsL2Rlc3RpbmF0aW9uLmdv) | `39.02% <0.00%> (ø)` | |
| [tunnel/msg/helper.go](https://codecov.io/gh/seqsense/aws-iot-device-sdk-go/pull/167/diff?src=pr&el=tree#diff-dHVubmVsL21zZy9oZWxwZXIuZ28=) | `60.00% <0.00%> (ø)` | |
| [tunnel/proxy.go](https://codecov.io/gh/seqsense/aws-iot-device-sdk-go/pull/167/diff?src=pr&el=tree#diff-dHVubmVsL3Byb3h5Lmdv) | `0.00% <0.00%> (ø)` | |
| [tunnel/server/api.go](https://codecov.io/gh/seqsense/aws-iot-device-sdk-go/pull/167/diff?src=pr&el=tree#diff-dHVubmVsL3NlcnZlci9hcGkuZ28=) | `35.29% <0.00%> (ø)` | |
| [tunnel/server/notifier.go](https://codecov.io/gh/seqsense/aws-iot-device-sdk-go/pull/167/diff?src=pr&el=tree#diff-dHVubmVsL3NlcnZlci9ub3RpZmllci5nbw==) | `0.00% <0.00%> (ø)` | |
| ... and [5 more](https://codecov.io/gh/seqsense/aws-iot-device-sdk-go/pull/167/diff?src=pr&el=tree-more) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/seqsense/aws-iot-device-sdk-go/pull/167?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/167?src=pr&el=footer). Last update [765016d...3f010dc](https://codecov.io/gh/seqsense/aws-iot-device-sdk-go/pull/167?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).


Patch:
diff --git a/awsiotdev.go b/awsiotdev.go
index 894f148..20cdefc 100644
--- a/awsiotdev.go
+++ b/awsiotdev.go
@@ -17,6 +17,8 @@ package awsiotdev
 
 import (
 	"github.com/at-wat/mqtt-go"
+
+	"github.com/seqsense/aws-iot-device-sdk-go/v4/internal/ioterr"
 )
 
 // Device is an AWS IoT device.
@@ -34,7 +36,7 @@ type device struct {
 func New(thingName string, dialer mqtt.Dialer, opts ...mqtt.ReconnectOption) (Device, error) {
 	cli, err := mqtt.NewReconnectClient(dialer, opts...)
 	if err != nil {
-		return nil, err
+		return nil, ioterr.New(err, "creating MQTT connection")
 	}
 	return &device{
 		Client:    cli,
diff --git a/dialer.go b/dialer.go
index faf39b0..d8e1a2c 100644
--- a/dialer.go
+++ b/dialer.go
@@ -20,6 +20,8 @@ import (
 	"github.com/aws/aws-sdk-go/aws/client"
 
 	"github.com/at-wat/mqtt-go"
+
+	"github.com/seqsense/aws-iot-device-sdk-go/v4/internal/ioterr"
 	presigner "github.com/seqsense/aws-iot-device-sdk-go/v4/presigner"
 )
 
@@ -39,14 +41,13 @@ func NewPresignDialer(sess client.ConfigProvider, endpoint string, opts ...mqtt.
 }
 
 func (d *presignDialer) Dial() (mqtt.ClientCloser, error) {
-	// Presign URL here.
 	url, err := d.signer.PresignWssNow(d.endpoint)
 	if err != nil {
-		return nil, err
+		return nil, ioterr.New(err, "presigning wss URL")
 	}
 	cli, err := mqtt.Dial(url, d.opts...)
 	if err != nil {
-		return nil, err
+		return nil, ioterr.New(err, "dialing")
 	}
 	return cli, nil
 }
@@ -56,7 +57,7 @@ func (d *presignDialer) Dial() (mqtt.ClientCloser, error) {
 func NewDialer(sess client.ConfigProvider, urlStr string, opts ...mqtt.DialOption) (mqtt.Dialer, error) {
 	u, err := url.Parse(urlStr)
 	if err != nil {
-		return nil, err
+		return nil, ioterr.New(err, "parsing server URL")
 	}
 	switch u.Scheme {
 	case "mqtts":
@@ -64,6 +65,6 @@ func NewDialer(sess client.ConfigProvider, urlStr string, opts ...mqtt.DialOptio
 	case "wss":
 		return NewPresignDialer(sess, u.Host)
 	default:
-		return nil, mqtt.ErrUnsupportedProtocol
+		return nil, ioterr.New(mqtt.ErrUnsupportedProtocol, "new dialer")
 	}
 }
diff --git a/go.mod b/go.mod
index def9e59..f28c837 100644
--- a/go.mod
+++ b/go.mod
@@ -3,6 +3,7 @@ module github.com/seqsense/aws-iot-device-sdk-go/v4
 require (
 	github.com/at-wat/mqtt-go v0.11.0
 	github.com/aws/aws-sdk-go v1.35.19
+	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/golang/protobuf v1.4.3
 	github.com/google/uuid v1.1.2
 	golang.org/x/net v0.0.0-20201031054903-ff519b6c9102
diff --git a/go.sum b/go.sum
index 0275356..c755e66 100644
--- a/go.sum
+++ b/go.sum
@@ -10,6 +10,8 @@ github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8
 github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
 github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
@@ -19,9 +21,7 @@ github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5y
 github.com/golang/protobuf v1.4.0-rc.1/go.mod h1:ceaxUfeHdC40wWswd/P6IGgMaK3YpKi5j83Wpe3EHw8=
 github.com/golang/protobuf v1.4.0-rc.1.0.20200221234624-67d41d38c208/go.mod h1:xKAWHe0F5eneWXFV3EuXVDTCmh+JuBKY0li0aMyXATA=
 github.com/golang/protobuf v1.4.0-rc.2/go.mod h1:LlEzMj4AhA7rCAGe4KMBDvJI+AwstrUpVNzEA03Pprs=
-github.com/golang/protobuf v1.4.0-rc.4.0.20200313231945-b860323f09d0 h1:aRz0NBceriICVtjhCgKkDvl+RudKu1CT6h0ZvUTrNfE=
 github.com/golang/protobuf v1.4.0-rc.4.0.20200313231945-b860323f09d0/go.mod h1:WU3c8KckQ9AFe+yFwt9sWVRKCVIyN9cPHBJSNnbL67w=
-github.com/golang/protobuf v1.4.0 h1:oOuy+ugB+P/kBdUnG5QaMXSIyJ1q38wWSojYCb3z5VQ=
 github.com/golang/protobuf v1.4.0/go.mod h1:jodUvKwWbYaEsadDk5Fwe5c77LiNKVO9IDvqG2KuDX0=
 github.com/golang/protobuf v1.4.1/go.mod h1:U8fpvMrcmy5pZrNK1lt4xCsGvpyWQ/VVv6QDs8UjoX8=
 github.com/golang/protobuf v1.4.3 h1:JjCZWpVbqXDqFVmTfYWEVTMIYrL/NPdPSCHPJ0T/raM=
@@ -29,7 +29,6 @@ github.com/golang/protobuf v1.4.3/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw
 github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
 github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
 github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
-github.com/google/go-cmp v0.4.0 h1:xsAVV57WRhGj6kEIi8ReJzQlHHqcBYCElAvkovg3B/4=
 github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.0 h1:/QaMHBdZ26BB3SSst0Iwl10Epc+xhTquomWX0oZEB6w=
 github.com/google/go-cmp v0.5.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
@@ -55,9 +54,7 @@ golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73r
 golang.org/x/net v0.0.0-20190213061140-3a22650c66bd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
-golang.org/x/net v0.0.0-20200202094626-16171245cfb2 h1:CCH4IOTTfewWjGOlSp+zGcjutRKlBEZQ6wTn8ozI/nI=
 golang.org/x/net v0.0.0-20200202094626-16171245cfb2/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
-golang.org/x/net v0.0.0-20200707034311-ab3426394381 h1:VXak5I6aEWmAXeQjA+QSZzlgNrpq9mjcfDemuexIKsU=
 golang.org/x/net v0.0.0-20200707034311-ab3426394381/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
 golang.org/x/net v0.0.0-20201031054903-ff519b6c9102 h1:42cLlJJdEh+ySyeUUbEQ5bsTiq8voBeTuweGVkY6Puw=
 golang.org/x/net v0.0.0-20201031054903-ff519b6c9102/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
@@ -70,7 +67,6 @@ golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5h
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/text v0.3.0 h1:g61tztE5qeGQ89tm6NTjjM9VPIm088od1l6aSorWRWg=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.3 h1:cokOdA+Jmi5PJGXLlLllQSgYigAEfHXJAERHVMaCc2k=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
@@ -93,10 +89,8 @@ google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLY
 google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0=
 google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM=
 google.golang.org/protobuf v1.20.1-0.20200309200217-e05f789c0967/go.mod h1:A+miEFZTKqfCUM6K7xSMQL9OKL/b6hQv+e19PK+JZNE=
-google.golang.org/protobuf v1.21.0 h1:qdOKuR/EIArgaWNjetjgTzgVTAZ+S/WXVrq9HW9zimw=
 google.golang.org/protobuf v1.21.0/go.mod h1:47Nbq4nVaFHyn7ilMalzfO3qCViNmqZ2kzikPIcrTAo=
 google.golang.org/protobuf v1.22.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
-google.golang.org/protobuf v1.23.0 h1:4MY060fB1DLGMB/7MBTLnwQUY6+F09GEiz6SsrNqyzM=
 google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
 google.golang.org/protobuf v1.23.1-0.20200526195155-81db48ad09cc/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
 google.golang.org/protobuf v1.25.0 h1:Ejskq+SyPohKW+1uil0JJMtmHCgJPJ/qWTxr8qp+R4c=
diff --git a/internal/ioterr/error.go b/internal/ioterr/error.go
new file mode 100644
index 0000000..da1bf40
--- /dev/null
+++ b/internal/ioterr/error.go
@@ -0,0 +1,69 @@
+// Copyright 2020 SEQSENSE, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package ioterr
+
+import (
+	"fmt"
+	"io"
+	"path/filepath"
+	"runtime"
+)
+
+// Error stores error for awsiotdev package.
+type Error struct {
+	Err     error
+	Failure string
+	File    string
+	Line    int
+}
+
+func (e *Error) Error() string {
+	return fmt.Sprintf("%s [%s:%d]: %s", e.Failure, filepath.Base(e.File), e.Line, e.Err.Error())
+}
+
+func (e *Error) Unwrap() error {
+	return e.Err
+}
+
+// New returns wrapped error.
+func New(err error, failure string) error {
+	return newImpl(err, failure)
+}
+
+// Newf returns wrapped error.
+func Newf(err error, format string, v ...interface{}) error {
+	return newImpl(err, fmt.Sprintf(format, v...))
+}
+
+func newImpl(err error, failure string) error {
+	switch err {
+	case io.EOF:
+		return io.EOF
+	case nil:
+		return nil
+	}
+
+	_, file, line, ok := runtime.Caller(2)
+	if !ok {
+		file = "unknown"
+		line = -1
+	}
+	return &Error{
+		Failure: failure,
+		Err:     err,
+		File:    file,
+		Line:    line,
+	}
+}
diff --git a/jobs/jobs.go b/jobs/jobs.go
index 43b13a6..5026d33 100644
--- a/jobs/jobs.go
+++ b/jobs/jobs.go
@@ -22,7 +22,9 @@ import (
 	"sync"
 
 	"github.com/at-wat/mqtt-go"
+
 	"github.com/seqsense/aws-iot-device-sdk-go/v4"
+	"github.com/seqsense/aws-iot-device-sdk-go/v4/internal/ioterr"
 )
 
 // Jobs is an interface of IoT Jobs.
@@ -80,7 +82,7 @@ func New(ctx context.Context, cli awsiotdev.Device) (Jobs, error) {
 		{j.topic("get/rejected"), mqtt.HandlerFunc(j.rejected)},
 	} {
 		if err := j.ServeMux.Handle(sub.topic, sub.handler); err != nil {
-			return nil, err
+			return nil, ioterr.New(err, "registering message handlers")
 		}
 	}
 
@@ -90,7 +92,7 @@ func New(ctx context.Context, cli awsiotdev.Device) (Jobs, error) {
 		mqtt.Subscription{Topic: j.topic("+/get/#"), QoS: mqtt.QoS1},
 	)
 	if err != nil {
-		return nil, err
+		return nil, ioterr.New(err, "subscribing jobs topics")
 	}
 	return j, nil
 }
@@ -98,7 +100,7 @@ func New(ctx context.Context, cli awsiotdev.Device) (Jobs, error) {
 func (j *jobs) notify(msg *mqtt.Message) {
 	m := &jobExecutionsChangedMessage{}
 	if err := json.Unmarshal(msg.Payload, m); err != nil {
-		j.handleError(err)
+		j.handleError(ioterr.New(err, "unmarshaling job executions changed message"))
 		return
 	}
 	j.mu.Lock()
@@ -124,7 +126,7 @@ func (j *jobs) GetPendingJobs(ctx context.Context) (map[JobExecutionState][]JobE
 
 	breq, err := json.Marshal(req)
 	if err != nil {
-		return nil, err
+		return nil, ioterr.New(err, "marshaling request")
 	}
 	if err := j.cli.Publish(ctx,
 		&mqtt.Message{
@@ -133,12 +135,12 @@ func (j *jobs) GetPendingJobs(ctx context.Context) (map[JobExecutionState][]JobE
 			Payload: breq,
 		},
 	); err != nil {
-		return nil, err
+		return nil, ioterr.New(err, "sending request")
 	}
 
 	select {
 	case <-ctx.Done():
-		return nil, ctx.Err()
+		return nil, ioterr.New(ctx.Err(), "getting pending job")
 	case res := <-ch:
 		switch r := res.(type) {
 		case *getPendingJobExecutionsResponse:
@@ -149,7 +151,7 @@ func (j *jobs) GetPendingJobs(ctx context.Context) (map[JobExecutionState][]JobE
 		case *ErrorResponse:
 			return nil, r
 		default:
-			return nil, ErrInvalidResponse
+			return nil, ioterr.New(ErrInvalidResponse, "getting pending job")
 		}
 	}
 }
@@ -171,7 +173,7 @@ func (j *jobs) DescribeJob(ctx context.Context, id string) (*JobExecution, error
 
 	breq, err := json.Marshal(req)
 	if err != nil {
-		return nil, err
+		return nil, ioterr.New(err, "marshaling request")
 	}
 	if err := j.cli.Publish(ctx,
 		&mqtt.Message{
@@ -180,12 +182,12 @@ func (j *jobs) DescribeJob(ctx context.Context, id string) (*JobExecution, error
 			Payload: breq,
 		},
 	); err != nil {
-		return nil, err
+		return nil, ioterr.New(err, "sending request")
 	}
 
 	select {
 	case <-ctx.Done():
-		return nil, ctx.Err()
+		return nil, ioterr.New(ctx.Err(), "describing job")
 	case res := <-ch:
 		switch r := res.(type) {
 		case *describeJobExecutionResponse:
@@ -193,7 +195,7 @@ func (j *jobs) DescribeJob(ctx context.Context, id string) (*JobExecution, error
 		case *ErrorResponse:
 			return nil, r
 		default:
-			return nil, ErrInvalidResponse
+			return nil, ioterr.New(ErrInvalidResponse, "describing job")
 		}
 	}
 }
@@ -224,7 +226,7 @@ func (j *jobs) UpdateJob(ctx context.Context, je *JobExecution, s JobExecutionSt
 
 	breq, err := json.Marshal(req)
 	if err != nil {
-		return err
+		return ioterr.New(err, "marshaling request")
 	}
 	if err := j.cli.Publish(ctx,
 		&mqtt.Message{
@@ -233,12 +235,12 @@ func (j *jobs) UpdateJob(ctx context.Context, je *JobExecution, s JobExecutionSt
 			Payload: breq,
 		},
 	); err != nil {
-		return err
+		return ioterr.New(err, "sending request")
 	}
 
 	select {
 	case <-ctx.Done():
-		return ctx.Err()
+		return ioterr.New(ctx.Err(), "updating job")
 	case res := <-ch:
 		switch r := res.(type) {
 		case *updateJobExecutionResponse:
@@ -246,7 +248,7 @@ func (j *jobs) UpdateJob(ctx context.Context, je *JobExecution, s JobExecutionSt
 		case *ErrorResponse:
 			return r
 		default:
-			return ErrInvalidResponse
+			return ioterr.New(ErrInvalidResponse, "updating job")
 		}
 	}
 }
@@ -271,7 +273,7 @@ func (j *jobs) handleResponse(r interface{}) {
 func (j *jobs) getAccepted(msg *mqtt.Message) {
 	res := &getPendingJobExecutionsResponse{}
 	if err := json.Unmarshal(msg.Payload, res); err != nil {
-		j.handleError(err)
+		j.handleError(ioterr.New(err, "unmarshaling pending job executions response"))
 		return
 	}
 	j.handleResponse(res)
@@ -280,7 +282,7 @@ func (j *jobs) getAccepted(msg *mqtt.Message) {
 func (j *jobs) getJobAccepted(msg *mqtt.Message) {
 	res := &describeJobExecutionResponse{}
 	if err := json.Unmarshal(msg.Payload, res); err != nil {
-		j.handleError(err)
+		j.handleError(ioterr.New(err, "unmarshaling describe job execution response"))
 		return
 	}
 	j.handleResponse(res)
@@ -289,7 +291,7 @@ func (j *jobs) getJobAccepted(msg *mqtt.Message) {
 func (j *jobs) updateJobAccepted(msg *mqtt.Message) {
 	res := &updateJobExecutionResponse{}
 	if err := json.Unmarshal(msg.Payload, res); err != nil {
-		j.handleError(err)
+		j.handleError(ioterr.New(err, "unmarshaling update job execution response"))
 		return
 	}
 	j.handleResponse(res)
@@ -298,7 +300,7 @@ func (j *jobs) updateJobAccepted(msg *mqtt.Message) {
 func (j *jobs) rejected(msg *mqtt.Message) {
 	e := &ErrorResponse{}
 	if err := json.Unmarshal(msg.Payload, e); err != nil {
-		j.handleError(err)
+		j.handleError(ioterr.New(err, "unmarshaling error response"))
 		return
 	}
 	j.handleResponse(e)
diff --git a/presigner/presign.go b/presigner/presign.go
index 7053189..73ec9ea 100644
--- a/presigner/presign.go
+++ b/presigner/presign.go
@@ -30,6 +30,8 @@ import (
 	"github.com/aws/aws-sdk-go/aws/client"
 	"github.com/aws/aws-sdk-go/aws/credentials"
 	"github.com/aws/aws-sdk-go/aws/signer/v4"
+
+	"github.com/seqsense/aws-iot-device-sdk-go/v4/internal/ioterr"
 )
 
 // Presigner is an AWS v4 signer wrapper for AWS IoT.
@@ -59,7 +61,7 @@ func (a *Presigner) PresignWss(endpoint string, expire time.Duration, from time.
 	}
 	cred, err := a.clientConfig.Config.Credentials.Get()
 	if err != nil {
-		return "", err
+		return "", ioterr.New(err, "getting credentials")
 	}
 	sessionToken := cred.SessionToken
 
@@ -71,7 +73,7 @@ func (a *Presigner) PresignWss(endpoint string, expire time.Duration, from time.
 
 	originalURL, err := url.Parse(fmt.Sprintf("wss://%s/mqtt", endpoint))
 	if err != nil {
-		return "", err
+		return "", ioterr.New(err, "parsing server URL")
 	}
 
 	req := &http.Request{
@@ -84,7 +86,7 @@ func (a *Presigner) PresignWss(endpoint string, expire time.Duration, from time.
 		expire, from,
 	)
 	if err != nil {
-		return "", err
+		return "", ioterr.New(err, "presigning URL")
 	}
 
 	ret := req.URL.String()
diff --git a/shadow/shadow.go b/shadow/shadow.go
index dfa3c2a..31c4eba 100644
--- a/shadow/shadow.go
+++ b/shadow/shadow.go
@@ -9,7 +9,9 @@ import (
 	"sync/atomic"
 
 	"github.com/at-wat/mqtt-go"
+
 	"github.com/seqsense/aws-iot-device-sdk-go/v4"
+	"github.com/seqsense/aws-iot-device-sdk-go/v4/internal/ioterr"
 )
 
 // Shadow is an interface of Thing Shadow.
@@ -80,7 +82,7 @@ func (s *shadow) handleResponse(r interface{}) {
 func (s *shadow) getAccepted(msg *mqtt.Message) {
 	doc := &ThingDocument{}
 	if err := json.Unmarshal(msg.Payload, doc); err != nil {
-		s.handleError(err)
+		s.handleError(ioterr.New(err, "unmarshaling thing document"))
 		return
 	}
 	s.mu.Lock()
@@ -94,7 +96,7 @@ func (s *shadow) getAccepted(msg *mqtt.Message) {
 func (s *shadow) rejected(msg *mqtt.Message) {
 	e := &ErrorResponse{}
 	if err := json.Unmarshal(msg.Payload, e); err != nil {
-		s.handleError(err)
+		s.handleError(ioterr.New(err, "unmarshaling error response"))
 		return
 	}
 	s.handleResponse(e)
@@ -103,14 +105,14 @@ func (s *shadow) rejected(msg *mqtt.Message) {
 func (s *shadow) updateAccepted(msg *mqtt.Message) {
 	doc := &thingDocumentRaw{}
 	if err := json.Unmarshal(msg.Payload, doc); err != nil {
-		s.handleError(err)
+		s.handleError(ioterr.New(err, "unmarshaling thing document"))
 		return
 	}
 	s.mu.Lock()
 	err := s.doc.update(doc)
 	s.mu.Unlock()
 	if err != nil {
-		s.handleError(err)
+		s.handleError(ioterr.New(err, "updating local thing document"))
 		return
 	}
 	s.handleResponse(doc)
@@ -119,7 +121,7 @@ func (s *shadow) updateAccepted(msg *mqtt.Message) {
 func (s *shadow) updateDelta(msg *mqtt.Message) {
 	state := &thingDelta{}
 	if err := json.Unmarshal(msg.Payload, state); err != nil {
-		s.handleError(err)
+		s.handleError(ioterr.New(err, "unmarshaling thing delta"))
 		return
 	}
 	s.mu.Lock()
@@ -134,7 +136,7 @@ func (s *shadow) updateDelta(msg *mqtt.Message) {
 func (s *shadow) deleteAccepted(msg *mqtt.Message) {
 	doc := &thingDocumentRaw{}
 	if err := json.Unmarshal(msg.Payload, doc); err != nil {
-		s.handleError(err)
+		s.handleError(ioterr.New(err, "unmarshaling thing document"))
 		return
 	}
 	s.mu.Lock()
@@ -171,7 +173,7 @@ func New(ctx context.Context, cli awsiotdev.Device) (Shadow, error) {
 		{s.topic("delete/rejected"), mqtt.HandlerFunc(s.rejected)},
 	} {
 		if err := s.ServeMux.Handle(sub.topic, sub.handler); err != nil {
-			return nil, err
+			return nil, ioterr.New(err, "registering message handlers")
 		}
 	}
 
@@ -185,7 +187,7 @@ func New(ctx context.Context, cli awsiotdev.Device) (Shadow, error) {
 		mqtt.Subscription{Topic: s.topic("delete/rejected"), QoS: mqtt.QoS1},
 	)
 	if err != nil {
-		return nil, err
+		return nil, ioterr.New(err, "subscribing shadow topics")
 	}
 	return s, nil
 }
@@ -193,7 +195,7 @@ func New(ctx context.Context, cli awsiotdev.Device) (Shadow, error) {
 func (s *shadow) Report(ctx context.Context, state interface{}) (*ThingDocument, error) {
 	rawState, err := json.Marshal(state)
 	if err != nil {
-		return nil, err
+		return nil, ioterr.New(err, "marshaling state")
 	}
 	token := s.token()
 	rawStateJSON := json.RawMessage(rawState)
@@ -202,7 +204,7 @@ func (s *shadow) Report(ctx context.Context, state interface{}) (*ThingDocument,
 		ClientToken: token,
 	})
 	if err != nil {
-		return nil, err
+		return nil, ioterr.New(err, "marshaling request")
 	}
 
 	ch := make(chan interface{}, 1)
@@ -220,12 +222,12 @@ func (s *shadow) Report(ctx context.Context, state interface{}) (*ThingDocument,
 		QoS:     mqtt.QoS1,
 		Payload: data,
 	}); err != nil {
-		return nil, err
+		return nil, ioterr.New(err, "sending request")
 	}
 
 	select {
 	case <-ctx.Done():
-		return nil, ctx.Err()
+		return nil, ioterr.New(ctx.Err(), "updating reported state")
 	case res := <-ch:
 		switch r := res.(type) {
 		case *thingDocumentRaw:
@@ -236,7 +238,7 @@ func (s *shadow) Report(ctx context.Context, state interface{}) (*ThingDocument,
 		case *ErrorResponse:
 			return nil, r
 		default:
-			return nil, ErrInvalidResponse
+			return nil, ioterr.New(ErrInvalidResponse, "updating reported state")
 		}
 	}
 }
@@ -244,7 +246,7 @@ func (s *shadow) Report(ctx context.Context, state interface{}) (*ThingDocument,
 func (s *shadow) Desire(ctx context.Context, state interface{}) (*ThingDocument, error) {
 	rawState, err := json.Marshal(state)
 	if err != nil {
-		return nil, err
+		return nil, ioterr.New(err, "marshaling state")
 	}
 	token := s.token()
 	rawStateJSON := json.RawMessage(rawState)
@@ -253,7 +255,7 @@ func (s *shadow) Desire(ctx context.Context, state interface{}) (*ThingDocument,
 		ClientToken: token,
 	})
 	if err != nil {
-		return nil, err
+		return nil, ioterr.New(err, "marshaling request")
 	}
 
 	ch := make(chan interface{}, 1)
@@ -271,12 +273,12 @@ func (s *shadow) Desire(ctx context.Context, state interface{}) (*ThingDocument,
 		QoS:     mqtt.QoS1,
 		Payload: data,
 	}); err != nil {
-		return nil, err
+		return nil, ioterr.New(err, "sending request")
 	}
 
 	select {
 	case <-ctx.Done():
-		return nil, ctx.Err()
+		return nil, ioterr.New(ctx.Err(), "updating desired state")
 	case res := <-ch:
 		switch r := res.(type) {
 		case *thingDocumentRaw:
@@ -287,7 +289,7 @@ func (s *shadow) Desire(ctx context.Context, state interface{}) (*ThingDocument,
 		case *ErrorResponse:
 			return nil, r
 		default:
-			return nil, ErrInvalidResponse
+			return nil, ioterr.New(ErrInvalidResponse, "updating desired state")
 		}
 	}
 }
@@ -298,7 +300,7 @@ func (s *shadow) Get(ctx context.Context) (*ThingDocument, error) {
 		ClientToken: token,
 	})
 	if err != nil {
-		return nil, err
+		return nil, ioterr.New(err, "marshaling requst")
 	}
 
 	ch := make(chan interface{}, 1)
@@ -316,12 +318,12 @@ func (s *shadow) Get(ctx context.Context) (*ThingDocument, error) {
 		QoS:     mqtt.QoS1,
 		Payload: []byte(data),
 	}); err != nil {
-		return nil, err
+		return nil, ioterr.New(err, "sending request")
 	}
 
 	select {
 	case <-ctx.Done():
-		return nil, ctx.Err()
+		return nil, ioterr.New(ctx.Err(), "getting document")
 	case res := <-ch:
 		switch r := res.(type) {
 		case *ThingDocument:
@@ -330,7 +332,7 @@ func (s *shadow) Get(ctx context.Context) (*ThingDocument, error) {
 		case *ErrorResponse:
 			return nil, r
 		default:
-			return nil, ErrInvalidResponse
+			return nil, ioterr.New(ErrInvalidResponse, "getting document")
 		}
 	}
 }
@@ -341,7 +343,7 @@ func (s *shadow) Delete(ctx context.Context) error {
 		ClientToken: token,
 	})
 	if err != nil {
-		return err
+		return ioterr.New(err, "marshaling request")
 	}
 
 	ch := make(chan interface{}, 1)
@@ -359,12 +361,12 @@ func (s *shadow) Delete(ctx context.Context) error {
 		QoS:     mqtt.QoS1,
 		Payload: []byte(data),
 	}); err != nil {
-		return err
+		return ioterr.New(err, "sending request")
 	}
 
 	select {
 	case <-ctx.Done():
-		return ctx.Err()
+		return ioterr.New(ctx.Err(), "deleting state")
 	case res := <-ch:
 		switch r := res.(type) {
 		case *thingDocumentRaw:
@@ -372,7 +374,7 @@ func (s *shadow) Delete(ctx context.Context) error {
 		case *ErrorResponse:
 			return r
 		default:
-			return ErrInvalidResponse
+			return ioterr.New(ErrInvalidResponse, "deleting state")
 		}
 	}
 }
diff --git a/shadow/state.go b/shadow/state.go
index ee3db8e..1fa87cc 100644
--- a/shadow/state.go
+++ b/shadow/state.go
@@ -4,6 +4,8 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+
+	"github.com/seqsense/aws-iot-device-sdk-go/v4/internal/ioterr"
 )
 
 // ErrVersionConflict means thing state update was aborted due to version conflict.
@@ -69,10 +71,10 @@ func (s *ThingDocument) update(state *thingDocumentRaw) error {
 	s.Version = state.Version
 	s.Timestamp = state.Timestamp
 	if err := updateStateRaw(s.State.Desired, state.State.Desired); err != nil {
-		return err
+		return ioterr.New(err, "updating desired state")
 	}
 	if err := updateStateRaw(s.State.Reported, state.State.Reported); err != nil {
-		return err
+		return ioterr.New(err, "updating reported state")
 	}
 	return nil
 }
@@ -100,7 +102,7 @@ func updateStateRaw(state map[string]interface{}, update json.RawMessage) error
 	}
 	var u map[string]interface{}
 	if err := json.Unmarshal([]byte(update), &u); err != nil {
-		return err
+		return ioterr.New(err, "unmarshaling update")
 	}
 	return updateState(state, u)
 }
@@ -117,7 +119,7 @@ func updateState(state map[string]interface{}, update map[string]interface{}) er
 		case map[string]interface{}:
 			if s, ok := state[key].(map[string]interface{}); ok {
 				if err := updateState(s, v); err != nil {
-					return err
+					return ioterr.New(err, "updating state")
 				}
 			} else {
 				state[key] = v
diff --git a/tunnel/destination.go b/tunnel/destination.go
index 9e96c0b..ae6f7bb 100644
--- a/tunnel/destination.go
+++ b/tunnel/destination.go
@@ -1,11 +1,25 @@
+// Copyright 2020 SEQSENSE, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package tunnel
 
 import (
-	"fmt"
 	"io"
 
 	"google.golang.org/protobuf/proto"
 
+	"github.com/seqsense/aws-iot-device-sdk-go/v4/internal/ioterr"
 	"github.com/seqsense/aws-iot-device-sdk-go/v4/tunnel/msg"
 )
 
@@ -18,7 +32,7 @@ func proxyDestination(ws io.ReadWriter, dialer Dialer, eh ErrorHandler) error {
 			if err == io.EOF {
 				return nil
 			}
-			return err
+			return ioterr.New(err, "reading length header")
 		}
 		l := int(sz[0])<<8 | int(sz[1])
 		if cap(b) < l {
@@ -29,12 +43,12 @@ func proxyDestination(ws io.ReadWriter, dialer Dialer, eh ErrorHandler) error {
 			if err == io.EOF {
 				return nil
 			}
-			return err
+			return ioterr.New(err, "reading message")
 		}
 		m := &msg.Message{}
 		if err := proto.Unmarshal(b, m); err != nil {
 			if eh != nil {
-				eh.HandleError(fmt.Errorf("unmarshal failed: %v", err))
+				eh.HandleError(ioterr.New(err, "unmarshaling message"))
 			}
 			continue
 		}
@@ -43,7 +57,7 @@ func proxyDestination(ws io.ReadWriter, dialer Dialer, eh ErrorHandler) error {
 			conn, err := dialer()
 			if err != nil {
 				if eh != nil {
-					eh.HandleError(fmt.Errorf("dial failed: %v", err))
+					eh.HandleError(ioterr.New(err, "dialing to destination"))
 				}
 				continue
 			}
@@ -67,7 +81,7 @@ func proxyDestination(ws io.ReadWriter, dialer Dialer, eh ErrorHandler) error {
 		case msg.Message_DATA:
 			if conn, ok := conns[m.StreamId]; ok {
 				if _, err := conn.Write(m.Payload); err != nil {
-					eh.HandleError(fmt.Errorf("write failed: %v", err))
+					eh.HandleError(ioterr.New(err, "writing message"))
 				}
 			}
 		}
diff --git a/tunnel/msg/helper.go b/tunnel/msg/helper.go
index 460eb2b..c249aaf 100644
--- a/tunnel/msg/helper.go
+++ b/tunnel/msg/helper.go
@@ -4,20 +4,22 @@ import (
 	"io"
 
 	"google.golang.org/protobuf/proto"
+
+	"github.com/seqsense/aws-iot-device-sdk-go/v4/internal/ioterr"
 )
 
 // WriteMessage marshals the message and sends to the writer.
 func WriteMessage(w io.Writer, m *Message) error {
 	bs, err := proto.Marshal(m)
 	if err != nil {
-		return err
+		return ioterr.New(err, "marshaling message")
 	}
 	l := len(bs)
 	b := append([]byte{byte(l >> 8), byte(l)}, bs...)
 	for p := 0; p < len(b); {
 		n, err := w.Write(b[p:])
 		if err != nil {
-			return err
+			return ioterr.New(err, "writing message")
 		}
 		p += n
 	}
diff --git a/tunnel/proxy.go b/tunnel/proxy.go
index 5158226..a913632 100644
--- a/tunnel/proxy.go
+++ b/tunnel/proxy.go
@@ -9,6 +9,8 @@ import (
 	"time"
 
 	"golang.org/x/net/websocket"
+
+	"github.com/seqsense/aws-iot-device-sdk-go/v4/internal/ioterr"
 )
 
 const (
@@ -31,7 +33,7 @@ type Dialer func() (io.ReadWriteCloser, error)
 func ProxyDestination(dialer Dialer, endpoint, token string, opts ...ProxyOption) error {
 	ws, opt, err := openProxyConn(endpoint, "destination", token, opts...)
 	if err != nil {
-		return err
+		return ioterr.New(err, "opening proxy destination")
 	}
 
 	pingCancel := newPinger(ws, opt.PingPeriod)
@@ -46,7 +48,7 @@ func ProxyDestination(dialer Dialer, endpoint, token string, opts ...ProxyOption
 func ProxySource(listener net.Listener, endpoint, token string, opts ...ProxyOption) error {
 	ws, opt, err := openProxyConn(endpoint, "source", token, opts...)
 	if err != nil {
-		return err
+		return ioterr.New(err, "opening proxy source")
 	}
 
 	pingCancel := newPinger(ws, opt.PingPeriod)
@@ -62,7 +64,7 @@ func openProxyConn(endpoint, mode, token string, opts ...ProxyOption) (*websocke
 	}
 	for _, o := range opts {
 		if err := o(opt); err != nil {
-			return nil, nil, err
+			return nil, nil, ioterr.New(err, "applying options")
 		}
 	}
 
@@ -71,7 +73,7 @@ func openProxyConn(endpoint, mode, token string, opts ...ProxyOption) (*websocke
 		fmt.Sprintf("https://%s", endpoint),
 	)
 	if err != nil {
-		return nil, nil, err
+		return nil, nil, ioterr.New(err, "creating ws config")
 	}
 	if opt.Scheme == "wss" {
 		wsc.TlsConfig = &tls.Config{
@@ -86,7 +88,7 @@ func openProxyConn(endpoint, mode, token string, opts ...ProxyOption) (*websocke
 	wsc.Protocol = append(wsc.Protocol, websocketProtocol)
 	ws, err := websocket.DialConfig(wsc)
 	if err != nil {
-		return nil, nil, err
+		return nil, nil, ioterr.New(err, "creating ws dial config")
 	}
 	ws.PayloadType = websocket.BinaryFrame
 
diff --git a/tunnel/server/api.go b/tunnel/server/api.go
index 367574b..f0d5675 100644
--- a/tunnel/server/api.go
+++ b/tunnel/server/api.go
@@ -12,6 +12,8 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/private/protocol/json/jsonutil"
 	ist "github.com/aws/aws-sdk-go/service/iotsecuretunneling"
+
+	"github.com/seqsense/aws-iot-device-sdk-go/v4/internal/ioterr"
 	"github.com/seqsense/aws-iot-device-sdk-go/v4/tunnel"
 )
 
@@ -23,7 +25,7 @@ type apiHandler struct {
 
 func (h *apiHandler) openTunnel(in *ist.OpenTunnelInput) (*ist.OpenTunnelOutput, error) {
 	if err := in.Validate(); err != nil {
-		return nil, err
+		return nil, ioterr.New(err, "validating request")
 	}
 	lifetime := 12 * time.Hour
 	if in.TimeoutConfig != nil && in.TimeoutConfig.MaxLifetimeTimeoutMinutes != nil {
@@ -32,12 +34,12 @@ func (h *apiHandler) openTunnel(in *ist.OpenTunnelInput) (*ist.OpenTunnelOutput,
 
 	r1, err := uuid.NewRandom()
 	if err != nil {
-		return nil, err
+		return nil, ioterr.New(err, "generating uuid")
 	}
 
 	r2, err := uuid.NewRandom()
 	if err != nil {
-		return nil, err
+		return nil, ioterr.New(err, "generating uuid")
 	}
 
 	ctx, cancel := context.WithTimeout(context.Background(), lifetime)
@@ -57,7 +59,7 @@ func (h *apiHandler) openTunnel(in *ist.OpenTunnelInput) (*ist.OpenTunnelOutput,
 
 	id, err := h.tunnelHandler.add(ti)
 	if err != nil {
-		return nil, err
+		return nil, ioterr.New(err, "adding tunnel")
 	}
 
 	if h.notifier != nil {
@@ -71,7 +73,7 @@ func (h *apiHandler) openTunnel(in *ist.OpenTunnelInput) (*ist.OpenTunnelOutput,
 				Services:          ti.services,
 			},
 		); err != nil {
-			return nil, err
+			return nil, ioterr.New(err, "notifying destination")
 		}
 	}
 
@@ -85,13 +87,13 @@ func (h *apiHandler) openTunnel(in *ist.OpenTunnelInput) (*ist.OpenTunnelOutput,
 
 func (h *apiHandler) closeTunnel(in *ist.CloseTunnelInput) (*ist.CloseTunnelOutput, error) {
 	if err := in.Validate(); err != nil {
-		return nil, err
+		return nil, ioterr.New(err, "validating request")
 	}
 
 	id := *in.TunnelId
 
 	if err := h.tunnelHandler.remove(id); err != nil {
-		return nil, err
+		return nil, ioterr.New(err, "removing tunnel")
 	}
 
 	return &ist.CloseTunnelOutput{}, nil
diff --git a/tunnel/server/notifier.go b/tunnel/server/notifier.go
index 71427c5..b9ae11e 100644
--- a/tunnel/server/notifier.go
+++ b/tunnel/server/notifier.go
@@ -7,6 +7,7 @@ import (
 
 	"github.com/at-wat/mqtt-go"
 
+	"github.com/seqsense/aws-iot-device-sdk-go/v4/internal/ioterr"
 	"github.com/seqsense/aws-iot-device-sdk-go/v4/tunnel"
 )
 
@@ -35,7 +36,7 @@ func NewNotifier(cli mqtt.Client, opts ...NotifierOption) *Notifier {
 func (n *Notifier) notify(ctx context.Context, thingName string, notify *tunnel.Notification) error {
 	b, err := json.Marshal(notify)
 	if err != nil {
-		return err
+		return ioterr.New(err, "marshaling notification")
 	}
 	return n.cli.Publish(ctx,
 		&mqtt.Message{
diff --git a/tunnel/source.go b/tunnel/source.go
index fcd6c22..f7efc64 100644
--- a/tunnel/source.go
+++ b/tunnel/source.go
@@ -1,13 +1,13 @@
 package tunnel
 
 import (
-	"fmt"
 	"io"
 	"net"
 	"sync"
 
 	"google.golang.org/protobuf/proto"
 
+	"github.com/seqsense/aws-iot-device-sdk-go/v4/internal/ioterr"
 	"github.com/seqsense/aws-iot-device-sdk-go/v4/tunnel/msg"
 )
 
@@ -19,6 +19,9 @@ func proxySource(ws io.ReadWriter, listener net.Listener, eh ErrorHandler) error
 		for {
 			conn, err := listener.Accept()
 			if err != nil {
+				if eh != nil {
+					eh.HandleError(ioterr.New(err, "accepting source connection"))
+				}
 				return
 			}
 
@@ -33,7 +36,7 @@ func proxySource(ws io.ReadWriter, listener net.Listener, eh ErrorHandler) error
 				StreamId: id,
 			}); err != nil {
 				if eh != nil {
-					eh.HandleError(fmt.Errorf("message send failed: %v", err))
+					eh.HandleError(ioterr.New(err, "sending message"))
 				}
 				continue
 			}
@@ -49,7 +52,7 @@ func proxySource(ws io.ReadWriter, listener net.Listener, eh ErrorHandler) error
 			if err == io.EOF {
 				return nil
 			}
-			return err
+			return ioterr.New(err, "reading length header")
 		}
 		l := int(sz[0])<<8 | int(sz[1])
 		if cap(b) < l {
@@ -60,12 +63,12 @@ func proxySource(ws io.ReadWriter, listener net.Listener, eh ErrorHandler) error
 			if err == io.EOF {
 				return nil
 			}
-			return err
+			return ioterr.New(err, "reading message")
 		}
 		m := &msg.Message{}
 		if err := proto.Unmarshal(b, m); err != nil {
 			if eh != nil {
-				eh.HandleError(fmt.Errorf("unmarshal failed: %v", err))
+				eh.HandleError(ioterr.New(err, "unmarshaling message"))
 			}
 			continue
 		}
@@ -93,7 +96,7 @@ func proxySource(ws io.ReadWriter, listener net.Listener, eh ErrorHandler) error
 			muConns.Unlock()
 			if ok {
 				if _, err := conn.Write(m.Payload); err != nil {
-					eh.HandleError(fmt.Errorf("write failed: %v", err))
+					eh.HandleError(ioterr.New(err, "writing message"))
 				}
 			}
 		}
diff --git a/tunnel/tunnel.go b/tunnel/tunnel.go
index e1c8aa6..ca8d1c5 100644
--- a/tunnel/tunnel.go
+++ b/tunnel/tunnel.go
@@ -3,11 +3,13 @@ package tunnel
 import (
 	"context"
 	"encoding/json"
-	"fmt"
+	"errors"
 	"sync"
 
 	"github.com/at-wat/mqtt-go"
+
 	"github.com/seqsense/aws-iot-device-sdk-go/v4"
+	"github.com/seqsense/aws-iot-device-sdk-go/v4/internal/ioterr"
 )
 
 // Tunnel is an interface of secure tunneling.
@@ -36,6 +38,9 @@ type Options struct {
 // Option is a type of functional options.
 type Option func(*Options) error
 
+// ErrInvalidClientMode indicate that the requested client mode is not valid for the tunnel.
+var ErrInvalidClientMode = errors.New("invalid client mode")
+
 func (t *tunnel) topic(operation string) string {
 	return "$aws/things/" + t.thingName + "/tunnels/" + operation
 }
@@ -52,19 +57,19 @@ func New(ctx context.Context, cli awsiotdev.Device, dialer map[string]Dialer, op
 	}
 	for _, o := range opts {
 		if err := o(t.opts); err != nil {
-			return nil, err
+			return nil, ioterr.New(err, "applying options")
 		}
 	}
 
 	if err := t.ServeMux.Handle(t.opts.TopicFunc("notify"), mqtt.HandlerFunc(t.notify)); err != nil {
-		return nil, err
+		return nil, ioterr.New(err, "registering message handler")
 	}
 
 	err := cli.Subscribe(ctx,
 		mqtt.Subscription{Topic: t.opts.TopicFunc("notify"), QoS: mqtt.QoS1},
 	)
 	if err != nil {
-		return nil, err
+		return nil, ioterr.New(err, "subscribing tunnel topic")
 	}
 	return t, nil
 }
@@ -72,11 +77,11 @@ func New(ctx context.Context, cli awsiotdev.Device, dialer map[string]Dialer, op
 func (t *tunnel) notify(msg *mqtt.Message) {
 	n := &Notification{}
 	if err := json.Unmarshal(msg.Payload, n); err != nil {
-		t.handleError(err)
+		t.handleError(ioterr.New(err, "unmarshaling notification"))
 		return
 	}
 	if n.ClientMode != Destination {
-		t.handleError(fmt.Errorf("unsupported client mode (%s)", n.ClientMode))
+		t.handleError(ioterr.Newf(ErrInvalidClientMode, "requested %s", n.ClientMode))
 		return
 	}
 	for _, srv := range n.Services {
@@ -89,7 +94,7 @@ func (t *tunnel) notify(msg *mqtt.Message) {
 					WithErrorHandler(ErrorHandlerFunc(t.handleError)),
 				)
 				if err != nil {
-					t.handleError(err)
+					t.handleError(ioterr.New(err, "creating proxy destination"))
 				}
 			}()
 		}


Test patch:
diff --git a/dialer_test.go b/dialer_test.go
index 0d2721e..39b5cd4 100644
--- a/dialer_test.go
+++ b/dialer_test.go
@@ -1,13 +1,32 @@
 package awsiotdev
 
 import (
+	"context"
+	"crypto/rand"
+	"crypto/rsa"
+	"crypto/tls"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"encoding/pem"
+	"errors"
+	"io/ioutil"
+	"math/big"
+	"net"
+	"net/http"
+	"net/url"
+	"os"
 	"reflect"
 	"testing"
+	"time"
+
+	"golang.org/x/net/websocket"
 
 	"github.com/at-wat/mqtt-go"
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/client"
+	"github.com/aws/aws-sdk-go/aws/session"
 
+	"github.com/seqsense/aws-iot-device-sdk-go/v4/internal/ioterr"
 	"github.com/seqsense/aws-iot-device-sdk-go/v4/presigner"
 )
 
@@ -18,37 +37,161 @@ func (*dummyConfigProvider) ClientConfig(serviceName string, cfgs ...*aws.Config
 }
 
 func TestNewDialer(t *testing.T) {
-	cases := map[string]struct {
-		url    string
-		dialer interface{}
-		err    error
-	}{
-		"MQTTs": {
-			url:    "mqtts://hoge.foo:1234",
-			dialer: &mqtt.URLDialer{URL: "mqtts://hoge.foo:1234"},
-		},
-		"WebSockets": {
-			url: "wss://hoge.foo:1234/ep",
-			dialer: &presignDialer{
-				signer:   &presigner.Presigner{},
-				endpoint: "hoge.foo:1234",
+	t.Run("ValidURL", func(t *testing.T) {
+		cases := map[string]struct {
+			url    string
+			dialer interface{}
+			err    error
+		}{
+			"MQTTs": {
+				url:    "mqtts://hoge.foo:1234",
+				dialer: &mqtt.URLDialer{URL: "mqtts://hoge.foo:1234"},
 			},
-		},
-		"UnknownProtocol": {
-			url: "unknown://hoge.foo:1234",
-			err: mqtt.ErrUnsupportedProtocol,
-		},
+			"WebSockets": {
+				url: "wss://hoge.foo:1234/ep",
+				dialer: &presignDialer{
+					signer:   &presigner.Presigner{},
+					endpoint: "hoge.foo:1234",
+				},
+			},
+			"UnknownProtocol": {
+				url: "unknown://hoge.foo:1234",
+				err: mqtt.ErrUnsupportedProtocol,
+			},
+		}
+		for name, c := range cases {
+			c := c
+			t.Run(name, func(t *testing.T) {
+				d, err := NewDialer(&dummyConfigProvider{}, c.url)
+				if !errors.Is(err, c.err) {
+					var ie *ioterr.Error
+					if !errors.As(err, &ie) {
+						t.Errorf("Expected error type: ioterr.Error, actual: %T", err)
+					}
+					t.Fatalf("Expected error: %v, got: %v", c.err, err)
+				}
+				if !reflect.DeepEqual(c.dialer, d) {
+					t.Errorf("Expected dialer: %v, got: %v", c.dialer, d)
+				}
+			})
+		}
+	})
+	t.Run("InvalidURL", func(t *testing.T) {
+		_, err := NewDialer(&dummyConfigProvider{}, ":aaa")
+		var ie *ioterr.Error
+		if !errors.As(err, &ie) {
+			t.Errorf("Expected error type: %T, actual: %T", ie, err)
+		}
+		var ue *url.Error
+		if !errors.As(err, &ue) {
+			t.Errorf("Expected error type: %T, actual: %T", ue, err)
+		}
+	})
+}
+
+func TestPresignDialer(t *testing.T) {
+	os.Clearenv()
+	os.Setenv("AWS_ACCESS_KEY_ID", "AKAAAAAAAAAAAAAAAAAA")
+	os.Setenv("AWS_SECRET_ACCESS_KEY", "1111111111111111111111111111111111111111")
+	os.Setenv("AWS_REGION", "world-1")
+
+	sess := session.Must(session.NewSession())
+	ps := presigner.New(sess)
+
+	ln, err := net.Listen("tcp", ":0")
+	if err != nil {
+		t.Fatal(err)
 	}
-	for name, c := range cases {
-		c := c
-		t.Run(name, func(t *testing.T) {
-			d, err := NewDialer(&dummyConfigProvider{}, c.url)
-			if err != c.err {
-				t.Fatalf("Expected error: %v, got: %v", c.err, err)
-			}
-			if !reflect.DeepEqual(c.dialer, d) {
-				t.Errorf("Expected dialer: %v, got: %v", c.dialer, d)
+	wsConfig, err := websocket.NewConfig("wss://", "wss://")
+	if err != nil {
+		t.Fatal(err)
+	}
+	wsSrv := websocket.Server{
+		Config:  *wsConfig,
+		Handler: func(c *websocket.Conn) {},
+	}
+	chAccept := make(chan bool, 1)
+	srv := &http.Server{
+		Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			if !<-chAccept {
+				http.NotFound(w, r)
+				return
 			}
-		})
+			wsSrv.ServeHTTP(w, r)
+		}),
+	}
+	cert, priv, err := generateSelfSignedCert()
+	if err != nil {
+		t.Fatal(err)
+	}
+	go srv.ServeTLS(ln, cert, priv)
+	defer srv.Shutdown(context.Background())
+
+	d := &presignDialer{
+		signer:   ps,
+		endpoint: ln.Addr().String(),
+		opts:     []mqtt.DialOption{mqtt.WithTLSConfig(&tls.Config{InsecureSkipVerify: true})},
+	}
+
+	t.Run("Success", func(t *testing.T) {
+		chAccept <- true
+		conn, err := d.Dial()
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		_ = conn.Close()
+	})
+	t.Run("Error", func(t *testing.T) {
+		chAccept <- false
+		conn, err := d.Dial()
+		if err == nil {
+			_ = conn.Close()
+			t.Fatalf("Dial should fail")
+		}
+		var ie *ioterr.Error
+		if !errors.As(err, &ie) {
+			t.Fatalf("Expected error type: %T, actual: %T", ie, err)
+		}
+	})
+}
+
+func generateSelfSignedCert() (string, string, error) {
+	priv, err := rsa.GenerateKey(rand.Reader, 2048)
+	if err != nil {
+		return "", "", err
+	}
+	template := x509.Certificate{
+		SerialNumber: big.NewInt(1),
+		Subject: pkix.Name{
+			Organization: []string{"Foo"},
+		},
+		NotBefore:             time.Now(),
+		NotAfter:              time.Now().Add(time.Hour),
+		KeyUsage:              x509.KeyUsageDigitalSignature | x509.KeyUsageKeyEncipherment,
+		ExtKeyUsage:           []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
+		BasicConstraintsValid: true,
+	}
+	der, err := x509.CreateCertificate(rand.Reader, &template, &template, &priv.PublicKey, priv)
+	if err != nil {
+		return "", "", err
+	}
+	certFile, err := ioutil.TempFile("", "*.pub")
+	if err != nil {
+		return "", "", err
+	}
+	privFile, err := ioutil.TempFile("", "*.key")
+	if err != nil {
+		return "", "", err
+	}
+	if err := pem.Encode(certFile, &pem.Block{Type: "CERTIFICATE", Bytes: der}); err != nil {
+		return "", "", err
+	}
+	privBytes, err := x509.MarshalPKCS8PrivateKey(priv)
+	if err != nil {
+		return "", "", err
+	}
+	if err := pem.Encode(privFile, &pem.Block{Type: "PRIVATE KEY", Bytes: privBytes}); err != nil {
+		return "", "", err
 	}
+	return certFile.Name(), privFile.Name(), nil
 }
diff --git a/internal/ioterr/error_test.go b/internal/ioterr/error_test.go
new file mode 100644
index 0000000..020148c
--- /dev/null
+++ b/internal/ioterr/error_test.go
@@ -0,0 +1,56 @@
+// Copyright 2020 SEQSENSE, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package ioterr
+
+import (
+	"errors"
+	"io"
+	"regexp"
+	"testing"
+)
+
+func TestError(t *testing.T) {
+	errBase := errors.New("an error")
+	errWrapped := New(errBase, "info")
+	errWrapped2 := Newf(errBase, "info")
+
+	errStrRegex := regexp.MustCompile(`^info \[error_test\.go:[0-9]+\]: an error$`)
+
+	if !errStrRegex.MatchString(errWrapped.Error()) {
+		t.Errorf("Expected error string regexp: '%s', got: '%s'", errStrRegex, errWrapped.Error())
+	}
+	if errors.Unwrap(errWrapped) != errBase {
+		t.Errorf("Expected unwrapped error: %v, got: %v", errBase, errors.Unwrap(errWrapped))
+	}
+
+	if !errStrRegex.MatchString(errWrapped2.Error()) {
+		t.Errorf("Expected error string regexp: '%s', got: '%s'", errStrRegex, errWrapped2.Error())
+	}
+	if errors.Unwrap(errWrapped2) != errBase {
+		t.Errorf("Expected unwrapped error: %v, got: %v", errBase, errors.Unwrap(errWrapped2))
+	}
+}
+
+func TestError_Nil(t *testing.T) {
+	if err := New(nil, "test"); err != nil {
+		t.Errorf("Nil error should be nil, got: %v", err)
+	}
+}
+
+func TestError_EOF(t *testing.T) {
+	if err := New(io.EOF, "test"); err != io.EOF {
+		t.Errorf("io.EOF error should be io.EOF, got: %v", err)
+	}
+}
diff --git a/jobs/jobs_test.go b/jobs/jobs_test.go
index fa84798..945ac38 100644
--- a/jobs/jobs_test.go
+++ b/jobs/jobs_test.go
@@ -24,8 +24,12 @@ import (
 
 	"github.com/at-wat/mqtt-go"
 	mockmqtt "github.com/at-wat/mqtt-go/mock"
+
+	"github.com/seqsense/aws-iot-device-sdk-go/v4/internal/ioterr"
 )
 
+var errPublish = errors.New("publish failure")
+
 type mockDevice struct {
 	*mockmqtt.Client
 }
@@ -34,6 +38,29 @@ func (d *mockDevice) ThingName() string {
 	return "test"
 }
 
+func TestNew(t *testing.T) {
+	errDummy := errors.New("dummy error")
+
+	t.Run("SubscribeError", func(t *testing.T) {
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+
+		cli := &mockDevice{&mockmqtt.Client{
+			SubscribeFn: func(ctx context.Context, subs ...mqtt.Subscription) error {
+				return errDummy
+			},
+		}}
+		_, err := New(ctx, cli)
+		var ie *ioterr.Error
+		if !errors.As(err, &ie) {
+			t.Errorf("Expected error type: %T, got: %T", ie, err)
+		}
+		if !errors.Is(err, errDummy) {
+			t.Errorf("Expected error: %v, got: %v", errDummy, err)
+		}
+	})
+}
+
 func TestNotify(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 	defer cancel()
@@ -80,10 +107,11 @@ func TestNotify(t *testing.T) {
 
 func TestGetPendingJobs(t *testing.T) {
 	testCases := map[string]struct {
-		response      interface{}
-		responseTopic string
-		expected      interface{}
-		err           error
+		publishFailure bool
+		response       interface{}
+		responseTopic  string
+		expected       interface{}
+		err            error
 	}{
 		"Success": {
 			response: &getPendingJobExecutionsResponse{
@@ -111,6 +139,10 @@ func TestGetPendingJobs(t *testing.T) {
 				Message: "Reason",
 			},
 		},
+		"PublishError": {
+			publishFailure: true,
+			err:            errPublish,
+		},
 	}
 
 	for name, testCase := range testCases {
@@ -122,6 +154,9 @@ func TestGetPendingJobs(t *testing.T) {
 			cli := &mockDevice{
 				Client: &mockmqtt.Client{
 					PublishFn: func(ctx context.Context, msg *mqtt.Message) error {
+						if testCase.publishFailure {
+							return errPublish
+						}
 						req := &simpleRequest{}
 						if err := json.Unmarshal(msg.Payload, req); err != nil {
 							t.Error(err)
@@ -154,7 +189,12 @@ func TestGetPendingJobs(t *testing.T) {
 			jbs, err := j.GetPendingJobs(ctx)
 			if err != nil {
 				setClientToken(err, "")
-				if !reflect.DeepEqual(testCase.err, err) {
+				var er *ErrorResponse
+				if errors.As(err, &er) {
+					if !reflect.DeepEqual(testCase.err, er) {
+						t.Fatalf("Expected error: %v, got: %v", testCase.err, err)
+					}
+				} else if !errors.Is(err, testCase.err) {
 					t.Fatalf("Expected error: %v, got: %v", testCase.err, err)
 				}
 			} else {
@@ -169,6 +209,7 @@ func TestGetPendingJobs(t *testing.T) {
 func TestDescribeJob(t *testing.T) {
 	testCases := map[string]struct {
 		id              string
+		publishFailure  bool
 		expectedRequest interface{}
 		response        interface{}
 		responseTopic   string
@@ -209,6 +250,14 @@ func TestDescribeJob(t *testing.T) {
 				Message: "Reason",
 			},
 		},
+		"PublishError": {
+			id:             "testID",
+			publishFailure: true,
+			expectedRequest: &describeJobExecutionRequest{
+				IncludeJobDocument: true,
+			},
+			err: errPublish,
+		},
 	}
 
 	for name, testCase := range testCases {
@@ -220,6 +269,9 @@ func TestDescribeJob(t *testing.T) {
 			cli := &mockDevice{
 				Client: &mockmqtt.Client{
 					PublishFn: func(ctx context.Context, msg *mqtt.Message) error {
+						if testCase.publishFailure {
+							return errPublish
+						}
 						req := &describeJobExecutionRequest{}
 						if err := json.Unmarshal(msg.Payload, req); err != nil {
 							t.Error(err)
@@ -259,7 +311,12 @@ func TestDescribeJob(t *testing.T) {
 			jb, err := j.DescribeJob(ctx, testCase.id)
 			if err != nil {
 				setClientToken(err, "")
-				if !reflect.DeepEqual(testCase.err, err) {
+				var er *ErrorResponse
+				if errors.As(err, &er) {
+					if !reflect.DeepEqual(testCase.err, er) {
+						t.Fatalf("Expected error: %v, got: %v", testCase.err, err)
+					}
+				} else if !errors.Is(err, testCase.err) {
 					t.Fatalf("Expected error: %v, got: %v", testCase.err, err)
 				}
 			} else {
@@ -273,6 +330,7 @@ func TestDescribeJob(t *testing.T) {
 
 func TestUpdateJob(t *testing.T) {
 	testCases := map[string]struct {
+		publishFailure  bool
 		execution       *JobExecution
 		status          JobExecutionState
 		options         []UpdateJobOption
@@ -380,6 +438,17 @@ func TestUpdateJob(t *testing.T) {
 				Message: "Reason",
 			},
 		},
+		"PublishError": {
+			publishFailure: true,
+			execution: &JobExecution{
+				JobID:         "testID",
+				JobDocument:   "doc",
+				StatusDetails: map[string]string{},
+				VersionNumber: 6,
+			},
+			status: Queued,
+			err:    errPublish,
+		},
 	}
 
 	for name, testCase := range testCases {
@@ -391,6 +460,9 @@ func TestUpdateJob(t *testing.T) {
 			cli := &mockDevice{
 				Client: &mockmqtt.Client{
 					PublishFn: func(ctx context.Context, msg *mqtt.Message) error {
+						if testCase.publishFailure {
+							return errPublish
+						}
 						req := &updateJobExecutionRequest{}
 						if err := json.Unmarshal(msg.Payload, req); err != nil {
 							t.Error(err)
@@ -430,10 +502,58 @@ func TestUpdateJob(t *testing.T) {
 			err = j.UpdateJob(ctx, testCase.execution, testCase.status, testCase.options...)
 			if err != nil {
 				setClientToken(err, "")
-				if !reflect.DeepEqual(testCase.err, err) {
+				var er *ErrorResponse
+				if errors.As(err, &er) {
+					if !reflect.DeepEqual(testCase.err, er) {
+						t.Fatalf("Expected error: %v, got: %v", testCase.err, err)
+					}
+				} else if !errors.Is(err, testCase.err) {
 					t.Fatalf("Expected error: %v, got: %v", testCase.err, err)
 				}
 			}
 		})
 	}
 }
+
+func TestHandlers_InvalidResponse(t *testing.T) {
+	for _, topic := range []string{
+		"notify",
+		"test/get/accepted",
+		"test/get/rejected",
+		"test/update/accepted",
+		"test/update/rejected",
+		"get/accepted",
+		"get/rejected",
+	} {
+		topic := topic
+		t.Run(topic, func(t *testing.T) {
+			ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+			defer cancel()
+
+			cli := &mockDevice{Client: &mockmqtt.Client{}}
+
+			j, err := New(ctx, cli)
+			if err != nil {
+				t.Fatal(err)
+			}
+			chErr := make(chan error, 1)
+			j.OnError(func(err error) { chErr <- err })
+			cli.Handle(j)
+
+			cli.Serve(&mqtt.Message{
+				Topic:   j.(*jobs).topic(topic),
+				Payload: []byte{0xff, 0xff, 0xff},
+			})
+
+			select {
+			case err := <-chErr:
+				var ie *ioterr.Error
+				if !errors.As(err, &ie) {
+					t.Errorf("Expected error type: %T, got: %T", ie, err)
+				}
+			case <-ctx.Done():
+				t.Fatal("Timeout")
+			}
+		})
+	}
+}
diff --git a/shadow/shadow_test.go b/shadow/shadow_test.go
index 40302f2..03be439 100644
--- a/shadow/shadow_test.go
+++ b/shadow/shadow_test.go
@@ -3,12 +3,15 @@ package shadow
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"reflect"
 	"testing"
 	"time"
 
-	mqtt "github.com/at-wat/mqtt-go"
+	"github.com/at-wat/mqtt-go"
 	mockmqtt "github.com/at-wat/mqtt-go/mock"
+
+	"github.com/seqsense/aws-iot-device-sdk-go/v4/internal/ioterr"
 )
 
 type mockDevice struct {
@@ -19,6 +22,29 @@ func (d *mockDevice) ThingName() string {
 	return "test"
 }
 
+func TestNew(t *testing.T) {
+	errDummy := errors.New("dummy error")
+
+	t.Run("SubscribeError", func(t *testing.T) {
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+
+		cli := &mockDevice{&mockmqtt.Client{
+			SubscribeFn: func(ctx context.Context, subs ...mqtt.Subscription) error {
+				return errDummy
+			},
+		}}
+		_, err := New(ctx, cli)
+		var ie *ioterr.Error
+		if !errors.As(err, &ie) {
+			t.Errorf("Expected error type: %T, got: %T", ie, err)
+		}
+		if !errors.Is(err, errDummy) {
+			t.Errorf("Expected error: %v, got: %v", errDummy, err)
+		}
+	})
+}
+
 func TestHandlers(t *testing.T) {
 	t.Run("Rejected", func(t *testing.T) {
 		chErr := make(chan error, 1)
@@ -48,8 +74,9 @@ func TestHandlers(t *testing.T) {
 		s.rejected(&mqtt.Message{})
 		select {
 		case err := <-chErr:
-			if err == nil {
-				t.Error("onError must be called with non-nil error")
+			var ie *ioterr.Error
+			if !errors.As(err, &ie) {
+				t.Errorf("Expected error type: %T, got: %T", ie, err)
 			}
 		default:
 			t.Fatal("Timeout")
@@ -295,6 +322,50 @@ func TestHandlers(t *testing.T) {
 	})
 }
 
+func TestHandlers_InvalidResponse(t *testing.T) {
+	for _, topic := range []string{
+		"update/delta",
+		"update/accepted",
+		"update/rejected",
+		"get/accepted",
+		"get/rejected",
+		"delete/accepted",
+		"delete/rejected",
+	} {
+		topic := topic
+		t.Run(topic, func(t *testing.T) {
+			ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+			defer cancel()
+
+			var cli *mockDevice
+			cli = &mockDevice{Client: &mockmqtt.Client{}}
+
+			s, err := New(ctx, cli)
+			if err != nil {
+				t.Fatal(err)
+			}
+			chErr := make(chan error, 1)
+			s.OnError(func(err error) { chErr <- err })
+			cli.Handle(s)
+
+			cli.Serve(&mqtt.Message{
+				Topic:   s.(*shadow).topic(topic),
+				Payload: []byte{0xff, 0xff, 0xff},
+			})
+
+			select {
+			case err := <-chErr:
+				var ie *ioterr.Error
+				if !errors.As(err, &ie) {
+					t.Errorf("Expected error type: %T, got: %T", ie, err)
+				}
+			case <-ctx.Done():
+				t.Fatal("Timeout")
+			}
+		})
+	}
+}
+
 func TestOnDelta(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 	defer cancel()
diff --git a/shadow/state_test.go b/shadow/state_test.go
index 5a8152e..5be8650 100644
--- a/shadow/state_test.go
+++ b/shadow/state_test.go
@@ -2,9 +2,12 @@ package shadow
 
 import (
 	"encoding/json"
+	"errors"
 	"reflect"
 	"strings"
 	"testing"
+
+	"github.com/seqsense/aws-iot-device-sdk-go/v4/internal/ioterr"
 )
 
 func TestThingDocument_update(t *testing.T) {
@@ -73,6 +76,44 @@ func TestThingDocument_update(t *testing.T) {
 			*expected2, *doc,
 		)
 	}
+
+	if err := doc.update(&thingDocumentRaw{
+		State:     thingStateRaw{Desired: json.RawMessage(`{"key": {"key": "value"}}`)},
+		Version:   6,
+		Timestamp: 12348,
+	}); err != nil {
+		t.Fatal(err)
+	}
+
+	expected3 := &ThingDocument{
+		State: ThingState{
+			Desired: map[string]interface{}{"key": map[string]interface{}{"key": "value"}},
+		},
+		Version:   6,
+		Timestamp: 12348,
+	}
+	if !reflect.DeepEqual(*expected3, *doc) {
+		t.Errorf(
+			"Document must be cleared by nil\nexpected: %v\ngot: %v",
+			*expected3, *doc,
+		)
+	}
+
+	t.Run("InvalidState", func(t *testing.T) {
+		err := doc.update(&thingDocumentRaw{
+			State:     thingStateRaw{Desired: json.RawMessage(`{"broken"}`)},
+			Version:   10,
+			Timestamp: 20000,
+		})
+		var ie *ioterr.Error
+		if !errors.As(err, &ie) {
+			t.Errorf("Expected error type: %T, got: %T", ie, err)
+		}
+		var je *json.SyntaxError
+		if !errors.As(err, &je) {
+			t.Errorf("Expected error type: %T, got: %T", je, err)
+		}
+	})
 }
 
 func TestErrorResponse(t *testing.T) {
diff --git a/tunnel/msg/helper_test.go b/tunnel/msg/helper_test.go
new file mode 100644
index 0000000..a7420f3
--- /dev/null
+++ b/tunnel/msg/helper_test.go
@@ -0,0 +1,47 @@
+package msg
+
+import (
+	"bytes"
+	"errors"
+	"testing"
+
+	"github.com/seqsense/aws-iot-device-sdk-go/v4/internal/ioterr"
+)
+
+func TestWriteMessage(t *testing.T) {
+	t.Run("Success", func(t *testing.T) {
+		m := &Message{Type: Message_STREAM_START}
+		buf := &bytes.Buffer{}
+
+		if err := WriteMessage(buf, m); err != nil {
+			t.Fatal(err)
+		}
+		expected := []byte{0x02}
+		if bytes.Equal(expected, buf.Bytes()) {
+			t.Errorf("Expected: %v, got: %v", expected, buf.Bytes())
+		}
+	})
+	t.Run("WriterError", func(t *testing.T) {
+		m := &Message{Type: Message_STREAM_START}
+
+		errDummy := errors.New("write error")
+		buf := writerFunc(func(b []byte) (int, error) {
+			return 0, errDummy
+		})
+
+		err := WriteMessage(buf, m)
+		if !errors.Is(err, errDummy) {
+			t.Errorf("Expected error: %v, got: %v", errDummy, err)
+		}
+		var ie *ioterr.Error
+		if !errors.As(err, &ie) {
+			t.Errorf("Expected error type: ioterr.Error, actual: %T", err)
+		}
+	})
+}
+
+type writerFunc func([]byte) (int, error)
+
+func (f writerFunc) Write(b []byte) (int, error) {
+	return f(b)
+}
diff --git a/tunnel/proxy_test.go b/tunnel/proxy_test.go
index 5acdc3d..b0bbc6a 100644
--- a/tunnel/proxy_test.go
+++ b/tunnel/proxy_test.go
@@ -1,3 +1,17 @@
+// Copyright 2020 SEQSENSE, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package tunnel
 
 import (
@@ -6,50 +20,352 @@ import (
 	"net"
 	"sync"
 	"testing"
+	"time"
 
 	"google.golang.org/protobuf/proto"
 
+	"github.com/seqsense/aws-iot-device-sdk-go/v4/internal/ioterr"
 	"github.com/seqsense/aws-iot-device-sdk-go/v4/tunnel/msg"
 )
 
+var (
+	errConnect = errors.New("conn error")
+)
+
 func TestProxyDestination(t *testing.T) {
-	tca, tcb := net.Pipe()
-	ca, cb := net.Pipe()
-
-	var wg sync.WaitGroup
-	wg.Add(1)
-	defer wg.Wait()
-
-	go func() {
-		defer wg.Done()
-		err := proxyDestination(tca,
-			func() (io.ReadWriteCloser, error) {
-				return cb, nil
+	t.Run("Success", func(t *testing.T) {
+		tca, tcb := net.Pipe()
+		ca, cb := net.Pipe()
+
+		var wg sync.WaitGroup
+		wg.Add(1)
+		defer wg.Wait()
+
+		go func() {
+			defer wg.Done()
+			err := proxyDestination(tca,
+				func() (io.ReadWriteCloser, error) { return cb, nil },
+				nil,
+			)
+			if err != nil {
+				t.Error(err)
+			}
+		}()
+
+		payload1 := "the payload 1"
+		payload2 := "the payload 2"
+
+		// Check source to destination
+		msgs := []*msg.Message{
+			{
+				Type:     msg.Message_STREAM_START,
+				StreamId: 1,
+			},
+			{
+				Type:     msg.Message_DATA,
+				StreamId: 1,
+				Payload:  []byte(payload1),
 			},
-			nil,
-		)
+		}
+		for _, m := range msgs {
+			b, err := proto.Marshal(m)
+			if err != nil {
+				t.Fatal(err)
+			}
+			l := len(b)
+			_, err = tcb.Write(append(
+				[]byte{byte(l >> 8), byte(l)},
+				b...,
+			))
+			if err != nil {
+				t.Fatal(err)
+			}
+		}
+		bRecv := make([]byte, 100)
+		n, err := ca.Read(bRecv)
 		if err != nil {
-			t.Error(err)
+			t.Fatal(err)
+		}
+		if string(bRecv[:n]) != payload1 {
+			t.Errorf("payload differs, expected: %s, got: %s", payload1, string(bRecv[:n]))
+		}
+
+		// Check destination to source
+		if _, err := ca.Write([]byte(payload2)); err != nil {
+			t.Fatal(err)
+		}
+		sz := make([]byte, 2)
+		if _, err := io.ReadFull(tcb, sz); err != nil {
+			t.Fatal(err)
+		}
+		bSent := make([]byte, int(sz[0])<<8|int(sz[1]))
+		if _, err := io.ReadFull(tcb, bSent); err != nil {
+			t.Fatal(err)
+		}
+		m := &msg.Message{}
+		if err := proto.Unmarshal(bSent, m); err != nil {
+			t.Fatalf("unmarshal failed: %v", err)
+		}
+		msgExpected := &msg.Message{
+			Type:     msg.Message_DATA,
+			StreamId: 1,
+			Payload:  []byte(payload2),
+		}
+		if !proto.Equal(msgExpected, m) {
+			t.Errorf("message differes, expected: %v, got: %v", msgExpected, m)
+		}
+
+		if err := tcb.Close(); err != nil {
+			t.Fatal(err)
+		}
+	})
+	t.Run("DialError", func(t *testing.T) {
+		tca, tcb := net.Pipe()
+
+		var wg sync.WaitGroup
+		defer wg.Wait()
+
+		chErr := make(chan error)
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			if err := proxyDestination(tca,
+				func() (io.ReadWriteCloser, error) { return nil, errConnect },
+				ErrorHandlerFunc(func(err error) {
+					chErr <- err
+				}),
+			); err != nil {
+				t.Error(err)
+			}
+		}()
+
+		b, err := proto.Marshal(&msg.Message{
+			Type:     msg.Message_STREAM_START,
+			StreamId: 1,
+		})
+		if err != nil {
+			t.Fatal(err)
 		}
-	}()
+		l := len(b)
+		if _, err = tcb.Write(append([]byte{byte(l >> 8), byte(l)}, b...)); err != nil {
+			t.Fatal(err)
+		}
+		tcb.Close()
 
-	payload1 := "the payload 1"
-	payload2 := "the payload 2"
+		select {
+		case <-time.After(time.Second):
+			t.Fatal("Timeout")
+		case err := <-chErr:
+			var ie *ioterr.Error
+			if !errors.As(err, &ie) {
+				t.Errorf("Expected error type: %T, got: %T", ie, err)
+			}
+			if !errors.Is(err, errConnect) {
+				t.Errorf("Expected error: %v, got: %v", errConnect, err)
+			}
+		}
+	})
+	t.Run("WriteError", func(t *testing.T) {
+		tca, tcb := net.Pipe()
+		ca, cb := net.Pipe()
 
-	// Check source to destination
-	msgs := []*msg.Message{
-		{
+		var wg sync.WaitGroup
+		defer wg.Wait()
+
+		chErr := make(chan error)
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			if err := proxyDestination(tca,
+				func() (io.ReadWriteCloser, error) { return cb, nil },
+				ErrorHandlerFunc(func(err error) {
+					chErr <- err
+				}),
+			); err != nil {
+				t.Error(err)
+			}
+		}()
+
+		b, err := proto.Marshal(&msg.Message{
 			Type:     msg.Message_STREAM_START,
 			StreamId: 1,
-		},
-		{
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+		l := len(b)
+		if _, err = tcb.Write(append([]byte{byte(l >> 8), byte(l)}, b...)); err != nil {
+			t.Fatal(err)
+		}
+		ca.Close()
+
+		b, err = proto.Marshal(&msg.Message{
+			Type:     msg.Message_DATA,
+			StreamId: 1,
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+		l = len(b)
+		if _, err = tcb.Write(append([]byte{byte(l >> 8), byte(l)}, b...)); err != nil {
+			t.Fatal(err)
+		}
+		tcb.Close()
+
+		select {
+		case <-time.After(time.Second):
+			t.Fatal("Timeout")
+		case err := <-chErr:
+			var ie *ioterr.Error
+			if !errors.As(err, &ie) {
+				t.Errorf("Expected error type: %T, got: %T", ie, err)
+			}
+			if !errors.Is(err, io.ErrClosedPipe) {
+				t.Errorf("Expected error: %v, got: %v", io.ErrClosedPipe, err)
+			}
+		}
+	})
+	t.Run("UnexpectedEOF", func(t *testing.T) {
+		packets := map[string][]byte{
+			"LengthHeader": []byte{0x00},
+			"Contents":     []byte{0x00, 0x10, 0x00},
+		}
+		for name, packet := range packets {
+			packet := packet
+			t.Run(name, func(t *testing.T) {
+				ca, cb := net.Pipe()
+
+				go func() {
+					if _, err := cb.Write(packet); err != nil {
+						t.Error(err)
+					}
+					if err := cb.Close(); err != nil {
+						t.Error(err)
+					}
+				}()
+
+				err := proxyDestination(ca, nil, nil)
+
+				var ie *ioterr.Error
+				if !errors.As(err, &ie) {
+					t.Errorf("Expected error type: %T, got: %T", ie, err)
+				}
+				if !errors.Is(err, io.ErrUnexpectedEOF) {
+					t.Errorf("Expected error: %v, got: %v", io.ErrUnexpectedEOF, err)
+				}
+			})
+		}
+	})
+	t.Run("InvalidProto", func(t *testing.T) {
+		ca, cb := net.Pipe()
+
+		packet := []byte{0, 3, 0xff, 0xff, 0xff}
+
+		chErr := make(chan error)
+		go func() {
+			if err := proxyDestination(ca, nil,
+				ErrorHandlerFunc(func(err error) { chErr <- err }),
+			); err != nil {
+				t.Error(err)
+			}
+		}()
+
+		if _, err := cb.Write(packet); err != nil {
+			t.Error(err)
+		}
+		if err := cb.Close(); err != nil {
+			t.Error(err)
+		}
+
+		select {
+		case <-time.After(time.Second):
+			t.Fatal("Timeout")
+		case err := <-chErr:
+			var ie *ioterr.Error
+			if !errors.As(err, &ie) {
+				t.Errorf("Expected error type: %T, got: %T", ie, err)
+			}
+			if !errors.Is(err, io.ErrUnexpectedEOF) {
+				t.Errorf("Expected error: %v, got: %v", io.ErrUnexpectedEOF, err)
+			}
+		}
+	})
+}
+
+func TestProxySource(t *testing.T) {
+	t.Run("Success", func(t *testing.T) {
+		tca, tcb := net.Pipe()
+		ca, cb := net.Pipe()
+
+		var wg sync.WaitGroup
+		defer wg.Wait()
+
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			var i int
+			if err := proxySource(tca,
+				acceptFunc(func() (net.Conn, error) {
+					if i > 0 {
+						return nil, errors.New("done")
+					}
+					i++
+					return cb, nil
+				}),
+				nil,
+			); err != nil {
+				t.Error(err)
+			}
+		}()
+
+		payload1 := "the payload 1"
+		payload2 := "the payload 2"
+
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			if _, err := ca.Write([]byte(payload1)); err != nil {
+				t.Error(err)
+			}
+		}()
+
+		// Check source to destination
+		msgsExpected := []*msg.Message{
+			{
+				Type:     msg.Message_STREAM_START,
+				StreamId: 1,
+			},
+			{
+				Type:     msg.Message_DATA,
+				StreamId: 1,
+				Payload:  []byte(payload1),
+			},
+		}
+		for _, me := range msgsExpected {
+			sz := make([]byte, 2)
+			if _, err := io.ReadFull(tcb, sz); err != nil {
+				t.Fatal(err)
+			}
+			bSent := make([]byte, int(sz[0])<<8|int(sz[1]))
+			if _, err := io.ReadFull(tcb, bSent); err != nil {
+				t.Fatal(err)
+			}
+			m := &msg.Message{}
+			if err := proto.Unmarshal(bSent, m); err != nil {
+				t.Fatalf("unmarshal failed: %v", err)
+			}
+			if !proto.Equal(me, m) {
+				t.Errorf("message differs, expected: %v, got: %v", me, m)
+			}
+		}
+
+		// Check destination to source
+		msg := &msg.Message{
 			Type:     msg.Message_DATA,
 			StreamId: 1,
-			Payload:  []byte(payload1),
-		},
-	}
-	for _, m := range msgs {
-		b, err := proto.Marshal(m)
+			Payload:  []byte(payload2),
+		}
+		b, err := proto.Marshal(msg)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -61,146 +377,210 @@ func TestProxyDestination(t *testing.T) {
 		if err != nil {
 			t.Fatal(err)
 		}
-	}
-	bRecv := make([]byte, 100)
-	n, err := ca.Read(bRecv)
-	if err != nil {
-		t.Fatal(err)
-	}
-	if string(bRecv[:n]) != payload1 {
-		t.Errorf("payload differs, expected: %s, got: %s", payload1, string(bRecv[:n]))
-	}
-
-	// Check destination to source
-	if _, err := ca.Write([]byte(payload2)); err != nil {
-		t.Fatal(err)
-	}
-	sz := make([]byte, 2)
-	if _, err := io.ReadFull(tcb, sz); err != nil {
-		t.Fatal(err)
-	}
-	bSent := make([]byte, int(sz[0])<<8|int(sz[1]))
-	if _, err := io.ReadFull(tcb, bSent); err != nil {
-		t.Fatal(err)
-	}
-	m := &msg.Message{}
-	if err := proto.Unmarshal(bSent, m); err != nil {
-		t.Fatalf("unmarshal failed: %v", err)
-	}
-	msgExpected := &msg.Message{
-		Type:     msg.Message_DATA,
-		StreamId: 1,
-		Payload:  []byte(payload2),
-	}
-	if !proto.Equal(msgExpected, m) {
-		t.Errorf("message differes, expected: %v, got: %v", msgExpected, m)
-	}
-
-	// Check EOF
-	if err := tcb.Close(); err != nil {
-		t.Fatal(err)
-	}
-}
 
-func TestProxySource(t *testing.T) {
-	tca, tcb := net.Pipe()
-	ca, cb := net.Pipe()
-
-	var wg sync.WaitGroup
-	defer wg.Wait()
-
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
-		var i int
-		err := proxySource(tca,
-			acceptFunc(func() (net.Conn, error) {
-				if i > 0 {
-					return nil, errors.New("done")
-				}
-				i++
-				return cb, nil
-			}),
-			nil,
-		)
+		bRecv := make([]byte, 100)
+		n, err := ca.Read(bRecv)
 		if err != nil {
-			t.Error(err)
+			t.Fatal(err)
+		}
+		if string(bRecv[:n]) != payload2 {
+			t.Errorf("payload differs, expected: %s, got: %s", payload2, string(bRecv[:n]))
 		}
-	}()
-
-	payload1 := "the payload 1"
-	payload2 := "the payload 2"
 
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
-		if _, err := ca.Write([]byte(payload1)); err != nil {
-			t.Error(err)
+		// Check EOF
+		if err := tcb.Close(); err != nil {
+			t.Fatal(err)
 		}
-	}()
+	})
+	t.Run("AcceptError", func(t *testing.T) {
+		tca, tcb := net.Pipe()
 
-	// Check source to destination
-	msgsExpected := []*msg.Message{
-		{
-			Type:     msg.Message_STREAM_START,
-			StreamId: 1,
-		},
-		{
+		var wg sync.WaitGroup
+		defer wg.Wait()
+
+		chErr := make(chan error)
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			if err := proxySource(tca,
+				acceptFunc(func() (net.Conn, error) { return nil, errConnect }),
+				ErrorHandlerFunc(func(err error) {
+					chErr <- err
+				}),
+			); err != nil {
+				t.Error(err)
+			}
+		}()
+
+		// Check destination to source
+		msg := &msg.Message{
 			Type:     msg.Message_DATA,
 			StreamId: 1,
-			Payload:  []byte(payload1),
-		},
-	}
-	for _, me := range msgsExpected {
-		sz := make([]byte, 2)
-		if _, err := io.ReadFull(tcb, sz); err != nil {
+			Payload:  []byte("test"),
+		}
+		b, err := proto.Marshal(msg)
+		if err != nil {
 			t.Fatal(err)
 		}
-		bSent := make([]byte, int(sz[0])<<8|int(sz[1]))
-		if _, err := io.ReadFull(tcb, bSent); err != nil {
+		l := len(b)
+		if _, err = tcb.Write(append([]byte{byte(l >> 8), byte(l)}, b...)); err != nil {
 			t.Fatal(err)
 		}
-		m := &msg.Message{}
-		if err := proto.Unmarshal(bSent, m); err != nil {
-			t.Fatalf("unmarshal failed: %v", err)
+
+		if err := tcb.Close(); err != nil {
+			t.Fatal(err)
+		}
+
+		select {
+		case <-time.After(time.Second):
+			t.Fatal("Timeout")
+		case err := <-chErr:
+			var ie *ioterr.Error
+			if !errors.As(err, &ie) {
+				t.Errorf("Expected error type: %T, got: %T", ie, err)
+			}
+			if !errors.Is(err, errConnect) {
+				t.Errorf("Expected error: %v, got: %v", errConnect, err)
+			}
+		}
+	})
+	t.Run("UnexpectedEOF", func(t *testing.T) {
+		packets := map[string][]byte{
+			"LengthHeader": []byte{0x00},
+			"Contents":     []byte{0x00, 0x10, 0x00},
+		}
+		for name, packet := range packets {
+			packet := packet
+			t.Run(name, func(t *testing.T) {
+				ca, cb := net.Pipe()
+
+				var wg sync.WaitGroup
+				wg.Add(1)
+				go func() {
+					defer wg.Done()
+					if _, err := cb.Write(packet); err != nil {
+						t.Error(err)
+					}
+					if err := cb.Close(); err != nil {
+						t.Error(err)
+					}
+				}()
+
+				err := proxySource(ca,
+					acceptFunc(func() (net.Conn, error) {
+						wg.Wait()
+						return nil, errConnect
+					}),
+					nil,
+				)
+
+				var ie *ioterr.Error
+				if !errors.As(err, &ie) {
+					t.Errorf("Expected error type: %T, got: %T", ie, err)
+				}
+				if !errors.Is(err, io.ErrUnexpectedEOF) {
+					t.Errorf("Expected error: %v, got: %v", io.ErrUnexpectedEOF, err)
+				}
+			})
+		}
+	})
+	t.Run("InvalidProto", func(t *testing.T) {
+		ca, cb := net.Pipe()
+
+		packet := []byte{0, 3, 0xff, 0xff, 0xff}
+
+		var wg sync.WaitGroup
+		wg.Add(1)
+
+		chErr := make(chan error)
+		go func() {
+			if err := proxySource(ca,
+				acceptFunc(func() (net.Conn, error) {
+					wg.Wait()
+					return nil, errConnect
+				}),
+				ErrorHandlerFunc(func(err error) { chErr <- err }),
+			); err != nil {
+				t.Error(err)
+			}
+		}()
+
+		if _, err := cb.Write(packet); err != nil {
+			t.Error(err)
+		}
+		if err := cb.Close(); err != nil {
+			t.Error(err)
+		}
+
+		select {
+		case <-time.After(time.Second):
+			t.Fatal("Timeout")
+		case err := <-chErr:
+			var ie *ioterr.Error
+			if !errors.As(err, &ie) {
+				t.Errorf("Expected error type: %T, got: %T", ie, err)
+			}
+			if !errors.Is(err, io.ErrUnexpectedEOF) {
+				t.Errorf("Expected error: %v, got: %v", io.ErrUnexpectedEOF, err)
+			}
+		}
+
+		wg.Done()
+	})
+	t.Run("WriteError", func(t *testing.T) {
+		tca, tcb := net.Pipe()
+		_, cb := net.Pipe()
+		defer cb.Close()
+
+		var wg sync.WaitGroup
+		defer wg.Wait()
+
+		chAccept := make(chan struct{})
+		chDone := make(chan struct{})
+		chErr := make(chan error)
+
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			var i int
+			if err := proxySource(tca,
+				acceptFunc(func() (net.Conn, error) {
+					if i > 0 {
+						<-chDone
+						return nil, errors.New("done")
+					}
+					i++
+					defer func() {
+						close(chAccept)
+					}()
+					return cb, nil
+				}),
+				ErrorHandlerFunc(func(err error) { chErr <- err }),
+			); err != nil {
+				t.Error(err)
+			}
+		}()
+
+		<-chAccept
+		if err := tcb.Close(); err != nil {
+			t.Fatal(err)
+		}
+
+		select {
+		case <-time.After(time.Second):
+			t.Fatal("Timeout")
+		case err := <-chErr:
+			var ie *ioterr.Error
+			if !errors.As(err, &ie) {
+				t.Errorf("Expected error type: %T, got: %T", ie, err)
+			}
+			if !errors.Is(err, io.ErrClosedPipe) {
+				t.Errorf("Expected error: %v, got: %v", io.ErrClosedPipe, err)
+			}
 		}
-		if !proto.Equal(me, m) {
-			t.Errorf("message differs, expected: %v, got: %v", me, m)
-		}
-	}
-
-	// Check destination to source
-	msg := &msg.Message{
-		Type:     msg.Message_DATA,
-		StreamId: 1,
-		Payload:  []byte(payload2),
-	}
-	b, err := proto.Marshal(msg)
-	if err != nil {
-		t.Fatal(err)
-	}
-	l := len(b)
-	_, err = tcb.Write(append(
-		[]byte{byte(l >> 8), byte(l)},
-		b...,
-	))
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	bRecv := make([]byte, 100)
-	n, err := ca.Read(bRecv)
-	if err != nil {
-		t.Fatal(err)
-	}
-	if string(bRecv[:n]) != payload2 {
-		t.Errorf("payload differs, expected: %s, got: %s", payload2, string(bRecv[:n]))
-	}
-
-	// Check EOF
-	if err := tcb.Close(); err != nil {
-		t.Fatal(err)
-	}
+		close(chDone)
+		<-chErr // Error on accept error
+	})
 }
 
 type acceptFunc func() (net.Conn, error)
diff --git a/tunnel/server/api_test.go b/tunnel/server/api_test.go
index 11a76e4..8c02cc5 100644
--- a/tunnel/server/api_test.go
+++ b/tunnel/server/api_test.go
@@ -1,6 +1,7 @@
 package server
 
 import (
+	"errors"
 	"fmt"
 	"net"
 	"net/http"
@@ -11,8 +12,11 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/credentials"
 	"github.com/aws/aws-sdk-go/aws/endpoints"
+	"github.com/aws/aws-sdk-go/aws/request"
 	"github.com/aws/aws-sdk-go/aws/session"
 	ist "github.com/aws/aws-sdk-go/service/iotsecuretunneling"
+
+	"github.com/seqsense/aws-iot-device-sdk-go/v4/internal/ioterr"
 )
 
 func TestAPI(t *testing.T) {
@@ -77,6 +81,32 @@ func TestAPI(t *testing.T) {
 	t.Logf("%v", out)
 }
 
+func TestAPI_Validate(t *testing.T) {
+	h := apiHandler{}
+
+	var err error
+	var ie *ioterr.Error
+	var re request.ErrInvalidParams
+
+	_, err = h.openTunnel(&ist.OpenTunnelInput{
+		Tags: []*ist.Tag{},
+	})
+	if !errors.As(err, &ie) {
+		t.Errorf("Expected error type: %T, got: %T", ie, err)
+	}
+	if !errors.As(err, &re) {
+		t.Errorf("Expected error type: %T, got: %T", re, err)
+	}
+
+	_, err = h.closeTunnel(&ist.CloseTunnelInput{})
+	if !errors.As(err, &ie) {
+		t.Errorf("Expected error type: %T, got: %T", ie, err)
+	}
+	if !errors.As(err, &re) {
+		t.Errorf("Expected error type: %T, got: %T", re, err)
+	}
+}
+
 func newEndpointForFunc(port int) endpoints.Resolver {
 	return endpoints.ResolverFunc(func(service, region string, opts ...func(*endpoints.Options)) (endpoints.ResolvedEndpoint, error) {
 		return endpoints.ResolvedEndpoint{
diff --git a/tunnel/tunnel_test.go b/tunnel/tunnel_test.go
new file mode 100644
index 0000000..09d92fc
--- /dev/null
+++ b/tunnel/tunnel_test.go
@@ -0,0 +1,169 @@
+package tunnel
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"testing"
+	"time"
+
+	mqtt "github.com/at-wat/mqtt-go"
+	mockmqtt "github.com/at-wat/mqtt-go/mock"
+
+	"github.com/seqsense/aws-iot-device-sdk-go/v4/internal/ioterr"
+)
+
+type mockDevice struct {
+	*mockmqtt.Client
+}
+
+func (d *mockDevice) ThingName() string {
+	return "test"
+}
+
+func TestNew(t *testing.T) {
+	errDummy := errors.New("dummy error")
+
+	t.Run("OptionError", func(t *testing.T) {
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+
+		cli := &mockDevice{&mockmqtt.Client{}}
+		_, err := New(ctx, cli, map[string]Dialer{}, func(opts *Options) error {
+			return errDummy
+		})
+		var ie *ioterr.Error
+		if !errors.As(err, &ie) {
+			t.Errorf("Expected error type: %T, got: %T", ie, err)
+		}
+		if !errors.Is(err, errDummy) {
+			t.Errorf("Expected error: %v, got: %v", errDummy, err)
+		}
+	})
+	t.Run("SubscribeError", func(t *testing.T) {
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+
+		cli := &mockDevice{&mockmqtt.Client{
+			SubscribeFn: func(ctx context.Context, subs ...mqtt.Subscription) error {
+				return errDummy
+			},
+		}}
+		_, err := New(ctx, cli, map[string]Dialer{})
+		var ie *ioterr.Error
+		if !errors.As(err, &ie) {
+			t.Errorf("Expected error type: %T, got: %T", ie, err)
+		}
+		if !errors.Is(err, errDummy) {
+			t.Errorf("Expected error: %v, got: %v", errDummy, err)
+		}
+	})
+	t.Run("InvalidTopicFunc", func(t *testing.T) {
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+
+		cli := &mockDevice{&mockmqtt.Client{}}
+		_, err := New(ctx, cli, map[string]Dialer{}, func(opts *Options) error {
+			opts.TopicFunc = func(operation string) string {
+				return "##"
+			}
+			return nil
+		})
+		var ie *ioterr.Error
+		if !errors.As(err, &ie) {
+			t.Errorf("Expected error type: %T, got: %T", ie, err)
+		}
+		if !errors.Is(err, mqtt.ErrInvalidTopicFilter) {
+			t.Errorf("Expected error: %v, got: %v", mqtt.ErrInvalidTopicFilter, err)
+		}
+	})
+
+}
+
+func TestHandlers(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+
+	cli := &mockDevice{&mockmqtt.Client{}}
+	tu, err := New(ctx, cli, map[string]Dialer{})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	chErr := make(chan error, 1)
+	tu.OnError(func(err error) {
+		chErr <- err
+	})
+	cli.Handle(tu)
+
+	t.Run("Notify", func(t *testing.T) {
+		t.Run("NonJSON", func(t *testing.T) {
+			cli.Serve(&mqtt.Message{
+				Topic:   "$aws/things/test/tunnels/notify",
+				Payload: []byte("aaaa{"),
+			})
+			select {
+			case <-ctx.Done():
+				t.Fatal("Timeout")
+			case err := <-chErr:
+				var ie *ioterr.Error
+				if !errors.As(err, &ie) {
+					t.Errorf("Expected error type: %T, got: %T", ie, err)
+				}
+				var je *json.SyntaxError
+				if !errors.As(err, &je) {
+					t.Errorf("Expected error type: %T, got: %T", je, err)
+				}
+			}
+		})
+		t.Run("NonDestination", func(t *testing.T) {
+			cli.Serve(&mqtt.Message{
+				Topic:   "$aws/things/test/tunnels/notify",
+				Payload: []byte(`{"ClientMode": "source"}`),
+			})
+			select {
+			case <-ctx.Done():
+				t.Fatal("Timeout")
+			case err := <-chErr:
+				var ie *ioterr.Error
+				if !errors.As(err, &ie) {
+					t.Errorf("Expected error type: %T, got: %T", ie, err)
+				}
+				if !errors.Is(err, ErrInvalidClientMode) {
+					t.Errorf("Expected error: %v, got: %v", ErrInvalidClientMode, err)
+				}
+			}
+		})
+	})
+}
+
+func TestHandlers_InvalidResponse(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancel()
+
+	var cli *mockDevice
+	cli = &mockDevice{Client: &mockmqtt.Client{}}
+
+	s, err := New(ctx, cli, map[string]Dialer{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	chErr := make(chan error, 1)
+	s.OnError(func(err error) { chErr <- err })
+	cli.Handle(s)
+
+	cli.Serve(&mqtt.Message{
+		Topic:   s.(*tunnel).topic("notify"),
+		Payload: []byte{0xff, 0xff, 0xff},
+	})
+
+	select {
+	case err := <-chErr:
+		var ie *ioterr.Error
+		if !errors.As(err, &ie) {
+			t.Errorf("Expected error type: %T, got: %T", ie, err)
+		}
+	case <-ctx.Done():
+		t.Fatal("Timeout")
+	}
+}