channel/existing_conn_impl.go at main · openziti/channel · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
package channel

import (
	"crypto/x509"
	"errors"
	"fmt"
	"github.com/openziti/identity"
	"net"
	"sync"
	"time"
)

type existingConnImpl struct {
	peer         net.Conn
	id           *identity.TokenId
	connectionId string
	headers      map[int32][]byte
	closeLock    sync.Mutex
	closed       bool
	readF        readFunction
	marshalF     marshalFunction
}

func (impl *existingConnImpl) GetLocalAddr() net.Addr {
	return impl.peer.LocalAddr()
}

func (impl *existingConnImpl) GetRemoteAddr() net.Addr {
	return impl.peer.RemoteAddr()
}

func (impl *existingConnImpl) SetWriteTimeout(duration time.Duration) error {
	return impl.peer.SetWriteDeadline(time.Now().Add(duration))
}

func (impl *existingConnImpl) SetWriteDeadline(deadline time.Time) error {
	return impl.peer.SetWriteDeadline(deadline)
}

func (impl *existingConnImpl) rxHello() (*Message, error) {
	msg, readF, marshallF, err := readHello(impl.peer)
	impl.readF = readF
	impl.marshalF = marshallF
	return msg, err
}

func (impl *existingConnImpl) Rx() (*Message, error) {
	if impl.closed {
		return nil, errors.New("underlay closed")
	}
	return impl.readF(impl.peer)
}

func (impl *existingConnImpl) Tx(m *Message) error {
	if impl.closed {
		return errors.New("underlay closed")
	}

	data, err := impl.marshalF(m)
	if err != nil {
		return err
	}

	_, err = impl.peer.Write(data)
	if err != nil {
		return err
	}

	return nil
}

func (impl *existingConnImpl) Id() string {
	return impl.id.Token
}

func (impl *existingConnImpl) Headers() map[int32][]byte {
	return impl.headers
}

func (impl *existingConnImpl) LogicalName() string {
	return "existing"
}

func (impl *existingConnImpl) ConnectionId() string {
	return impl.connectionId
}

func (impl *existingConnImpl) Certificates() []*x509.Certificate {
	return nil
}

func (impl *existingConnImpl) Label() string {
	return fmt.Sprintf("u{%s}->i{%s}", impl.LogicalName(), impl.ConnectionId())
}

func (impl *existingConnImpl) Close() error {
	impl.closeLock.Lock()
	defer impl.closeLock.Unlock()

	if !impl.closed {
		impl.closed = true
		return impl.peer.Close()
	}
	return nil
}

func (impl *existingConnImpl) IsClosed() bool {
	return impl.closed
}

func newExistingImpl(peer net.Conn, version uint32) *existingConnImpl {
	readF := ReadV2
	marshalF := MarshalV2

	switch version {
	case 2:
		readF = ReadV2
		marshalF = MarshalV2
	case 3:
		readF = ReadV2
		marshalF = marshalV3
	}

	return &existingConnImpl{
		peer:     peer,
		readF:    readF,
		marshalF: marshalF,
	}
}