/
dilaer.go
139 lines (123 loc) · 2.92 KB
/
dilaer.go
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
130
131
132
133
134
135
136
137
138
139
package dialer
import (
"fmt"
"io"
"sync/atomic"
"github.com/Centny/gwf/util"
)
type Pipable interface {
Pipe(r io.ReadWriteCloser) error
}
type Conn interface {
Pipable
io.ReadWriteCloser
}
type CopyPipable struct {
io.ReadWriteCloser
piped uint32
}
func NewCopyPipable(raw io.ReadWriteCloser) *CopyPipable {
return &CopyPipable{ReadWriteCloser: raw}
}
func (c *CopyPipable) Pipe(r io.ReadWriteCloser) (err error) {
if atomic.CompareAndSwapUint32(&c.piped, 0, 1) {
go c.copyAndClose(c, r)
go c.copyAndClose(r, c)
} else {
err = fmt.Errorf("piped")
}
return
}
func (c *CopyPipable) copyAndClose(src io.ReadWriteCloser, dst io.ReadWriteCloser) {
io.Copy(dst, src)
dst.Close()
src.Close()
}
// Dialer is the interface that wraps the dialer
type Dialer interface {
Name() string
//initial dialer
Bootstrap(options util.Map) error
//
Options() util.Map
//match uri
Matched(uri string) bool
//dial raw connection
Dial(sid uint64, uri string, raw io.ReadWriteCloser) (r Conn, err error)
}
//Pool is the set of Dialer
type Pool struct {
Dialers []Dialer
}
//NewPool will return new Pool
func NewPool() (pool *Pool) {
pool = &Pool{}
return
}
//AddDialer will append dialer which is bootstraped to pool
func (p *Pool) AddDialer(dialers ...Dialer) (err error) {
p.Dialers = append(p.Dialers, dialers...)
return
}
func (p *Pool) Bootstrap(options util.Map) error {
dialerOptions := options.AryMapVal("dialers")
for _, option := range dialerOptions {
dtype := option.StrVal("type")
dialer := NewDialer(dtype)
if dialer == nil {
return fmt.Errorf("create dialer fail by %v", util.S2Json(option))
}
err := dialer.Bootstrap(option)
if err != nil {
return err
}
p.Dialers = append(p.Dialers, dialer)
}
if options.IntValV("standard", 0) > 0 {
p.Dialers = append(p.Dialers, NewCmdDialer(), NewEchoDialer(),
NewWebDialer(), NewTCPDialer())
} else {
if options.IntValV("cmd", 0) > 0 {
p.Dialers = append(p.Dialers, NewCmdDialer())
}
if options.IntValV("echo", 0) > 0 {
p.Dialers = append(p.Dialers, NewEchoDialer())
}
if options.IntValV("web", 0) > 0 {
p.Dialers = append(p.Dialers, NewWebDialer())
}
if options.IntValV("tcp", 0) > 0 {
p.Dialers = append(p.Dialers, NewTCPDialer())
}
}
return nil
}
//Dial the uri by dialer poo
func (p *Pool) Dial(sid uint64, uri string, pipe io.ReadWriteCloser) (r Conn, err error) {
for _, dialer := range p.Dialers {
if dialer.Matched(uri) {
r, err = dialer.Dial(sid, uri, pipe)
return
}
}
err = fmt.Errorf("uri(%v) is not supported(not matched dialer)", uri)
return
}
func DefaultDialerCreator(t string) (dialer Dialer) {
switch t {
case "balance":
dialer = NewBalancedDialer()
case "cmd":
dialer = NewCmdDialer()
case "echo":
dialer = NewEchoDialer()
case "socks":
dialer = NewSocksProxyDialer()
case "tcp":
dialer = NewTCPDialer()
case "web":
dialer = NewWebDialer()
}
return
}
var NewDialer = DefaultDialerCreator