4076575789

Build Type

push

github

Committed by Emmanuel BOUTON

Commit Message

feat: add descriptions to command status and enhance error message on binding error

Run Details

4 of 4 new or added lines in 1 file covered. (100.0%)

419 of 481 relevant lines covered (87.11%)

24.95 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

91.74

/transmittable.go

package gosmpp

import (
        "fmt"
        "net"
        "runtime"
        "sync"
        "sync/atomic"
        "time"

        "github.com/linxGnu/gosmpp/pdu"
)

var (
        // ErrConnectionClosing indicates transmitter is closing. Can not send any PDU.
        ErrConnectionClosing = fmt.Errorf("connection is closing, can not send PDU to SMSC")
)

type transmittable struct {
        settings Settings

        wg    sync.WaitGroup
        input chan pdu.PDU

        conn *Connection

        aliveState   int32
        pendingWrite int32
}

func newTransmittable(conn *Connection, settings Settings) *transmittable {
        t := &transmittable{
                settings:     settings,
                conn:         conn,
                input:        make(chan pdu.PDU, 1),
                aliveState:   Alive,
                pendingWrite: 0,
        }

        return t
}

func (t *transmittable) close(state State) (err error) {
        if atomic.CompareAndSwapInt32(&t.aliveState, Alive, Closed) {
                for atomic.LoadInt32(&t.pendingWrite) != 0 {
                        runtime.Gosched()
                }

                // notify daemon
                close(t.input)

                // wait daemon
                t.wg.Wait()

                // try to send unbind
                _, _ = t.write(pdu.NewUnbind())

                // close connection
                if state != StoppingProcessOnly {
                        err = t.conn.Close()
                }

                // notify transmitter closed
                if t.settings.OnClosed != nil {
                        t.settings.OnClosed(state)
                }
        }

        return
}

func (t *transmittable) closing(state State) {
        go func() {
                _ = t.close(state)
        }()
}

// Submit a PDU.
func (t *transmittable) Submit(p pdu.PDU) (err error) {
        atomic.AddInt32(&t.pendingWrite, 1)

        if atomic.LoadInt32(&t.aliveState) == Alive {
                t.input <- p
        } else {
                err = ErrConnectionClosing
        }

        atomic.AddInt32(&t.pendingWrite, -1)
        return
}

func (t *transmittable) start() {
        t.wg.Add(1)
        if t.settings.EnquireLink > 0 {
                go func() {
                        t.loopWithEnquireLink()
                        t.wg.Done()
                }()
        } else {
                go func() {
                        t.loop()
                        t.wg.Done()
                }()
        }
}

func (t *transmittable) drain() {
        for range t.input {
        }
}

func (t *transmittable) loop() {
        defer t.drain()

        for p := range t.input {
                if p != nil {
                        n, err := t.write(p)
                        if t.check(p, n, err) {
                                return
                        }
                }
        }
}

func (t *transmittable) loopWithEnquireLink() {
        ticker := time.NewTicker(t.settings.EnquireLink)
        defer func() {
                ticker.Stop()
                t.drain()
        }()

        for {
                select {
                case <-ticker.C:
                        eqp := pdu.NewEnquireLink()
                        n, err := t.write(eqp)
                        if t.check(eqp, n, err) {
                                return
                        }

                case p, ok := <-t.input:
                        if !ok {
                                return
                        }

                        if p != nil {
                                n, err := t.write(p)
                                if t.check(p, n, err) {
                                        return
                                }
                        }
                }
        }
}

// check error and do closing if need
func (t *transmittable) check(p pdu.PDU, n int, err error) (closing bool) {
        if err == nil {
                return
        }

        if t.settings.OnSubmitError != nil {
                t.settings.OnSubmitError(p, err)
        }

        if n == 0 {
                if nErr, ok := err.(net.Error); ok {
                        closing = nErr.Timeout()
                } else {
                        closing = true
                }
        } else {
                closing = true // force closing
        }

        if closing {
                t.closing(ConnectionIssue) // start closing
        }

        return
}

// low level writing
func (t *transmittable) write(p pdu.PDU) (n int, err error) {
        if t.settings.WriteTimeout > 0 {
                err = t.conn.SetWriteTimeout(t.settings.WriteTimeout)
        }

        if err == nil {
                n, err = t.conn.WritePDU(p)
        }

        return
}

1	package gosmpp
2
3	import (
4	"fmt"
5	"net"
6	"runtime"
7	"sync"
8	"sync/atomic"
9	"time"
10
11	"github.com/linxGnu/gosmpp/pdu"
12	)
13
14	var (
15	// ErrConnectionClosing indicates transmitter is closing. Can not send any PDU.
16	ErrConnectionClosing = fmt.Errorf("connection is closing, can not send PDU to SMSC")
17	)
18
19	type transmittable struct {
20	settings Settings
21
22	wg sync.WaitGroup
23	input chan pdu.PDU
24
25	conn *Connection
26
27	aliveState int32
28	pendingWrite int32
29	}
30
31	func newTransmittable(conn Connection, settings Settings) transmittable {	11✔
32	t := &transmittable{	11✔
33	settings: settings,	11✔
34	conn: conn,	11✔
35	input: make(chan pdu.PDU, 1),	11✔
36	aliveState: Alive,	11✔
37	pendingWrite: 0,	11✔
38	}	11✔
39		11✔
40	return t	11✔
41	}	11✔
42
43	func (t *transmittable) close(state State) (err error) {	17✔
44	if atomic.CompareAndSwapInt32(&t.aliveState, Alive, Closed) {	30✔
45	for atomic.LoadInt32(&t.pendingWrite) != 0 {	13✔
46	runtime.Gosched()	×
47	}	×
48
49	// notify daemon
50	close(t.input)	13✔
51		13✔
52	// wait daemon	13✔
53	t.wg.Wait()	13✔
54		13✔
55	// try to send unbind	13✔
56	_, _ = t.write(pdu.NewUnbind())	13✔
57		13✔
58	// close connection	13✔
59	if state != StoppingProcessOnly {	19✔
60	err = t.conn.Close()	6✔
61	}	6✔
62
63	// notify transmitter closed
64	if t.settings.OnClosed != nil {	26✔
65	t.settings.OnClosed(state)	13✔
66	}	13✔
67	}
68
69	return	17✔
70	}
71
72	func (t *transmittable) closing(state State) {	2✔
73	go func() {	4✔
74	_ = t.close(state)	2✔
75	}()	2✔
76	}
77
78	// Submit a PDU.
79	func (t *transmittable) Submit(p pdu.PDU) (err error) {	147✔
80	atomic.AddInt32(&t.pendingWrite, 1)	147✔
81		147✔
82	if atomic.LoadInt32(&t.aliveState) == Alive {	205✔
83	t.input <- p	58✔
84	} else {	147✔
85	err = ErrConnectionClosing	89✔
86	}	89✔
87
88	atomic.AddInt32(&t.pendingWrite, -1)	147✔
89	return	147✔
90	}
91
92	func (t *transmittable) start() {	11✔
93	t.wg.Add(1)	11✔
94	if t.settings.EnquireLink > 0 {	17✔
95	go func() {	12✔
96	t.loopWithEnquireLink()	6✔
97	t.wg.Done()	6✔
98	}()	6✔
99	} else {	5✔
100	go func() {	10✔
101	t.loop()	5✔
102	t.wg.Done()	5✔
103	}()	5✔
104	}
105	}
106
107	func (t *transmittable) drain() {	11✔
108	for range t.input {	11✔
109	}	×
110	}
111
112	func (t *transmittable) loop() {	5✔
113	defer t.drain()	5✔
114		5✔
115	for p := range t.input {	24✔
116	if p != nil {	38✔
117	n, err := t.write(p)	19✔
118	if t.check(p, n, err) {	19✔
119	return	×
120	}	×
121	}
122	}
123	}
124
125	func (t *transmittable) loopWithEnquireLink() {	6✔
126	ticker := time.NewTicker(t.settings.EnquireLink)	6✔
127	defer func() {	12✔
128	ticker.Stop()	6✔
129	t.drain()	6✔
130	}()	6✔
131
132	for {	85✔
133	select {	79✔
134	case <-ticker.C:	35✔
135	eqp := pdu.NewEnquireLink()	35✔
136	n, err := t.write(eqp)	35✔
137	if t.check(eqp, n, err) {	35✔
138	return	×
139	}	×
140
141	case p, ok := <-t.input:	44✔
142	if !ok {	50✔
143	return	6✔
144	}	6✔
145
146	if p != nil {	76✔
147	n, err := t.write(p)	38✔
148	if t.check(p, n, err) {	38✔
149	return	×
150	}	×
151	}
152	}
153	}
154	}
155
156	// check error and do closing if need
157	func (t *transmittable) check(p pdu.PDU, n int, err error) (closing bool) {	94✔
158	if err == nil {	186✔
159	return	92✔
160	}	92✔
161
162	if t.settings.OnSubmitError != nil {	4✔
163	t.settings.OnSubmitError(p, err)	2✔
164	}	2✔
165
166	if n == 0 {	3✔
167	if nErr, ok := err.(net.Error); ok {	1✔
168	closing = nErr.Timeout()	×
169	} else {	1✔
170	closing = true	1✔
171	}	1✔
172	} else {	1✔
173	closing = true // force closing	1✔
174	}	1✔
175
176	if closing {	4✔
177	t.closing(ConnectionIssue) // start closing	2✔
178	}	2✔
179
180	return	2✔
181	}
182
183	// low level writing
184	func (t *transmittable) write(p pdu.PDU) (n int, err error) {	107✔
185	if t.settings.WriteTimeout > 0 {	186✔
186	err = t.conn.SetWriteTimeout(t.settings.WriteTimeout)	79✔
187	}	79✔
188
189	if err == nil {	210✔
190	n, err = t.conn.WritePDU(p)	103✔
191	}	103✔
192
193	return	107✔
194	}

linxGnu / gosmpp / 4076575789

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous