vitalif
/
etcd
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Merge pull request #9571 from gyuho/net 

pkg/proxy: make/simplify interface more extensible
release-3.4
Gyuho Lee 2018-04-14 11:57:01 -07:00 committed by GitHub
parent 03cf9c45f8 5db4df762b
commit 9886e2161d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 666 additions and 601 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1157
pkg/proxy/server.go
View File

File diff suppressed because it is too large Load Diff

110
pkg/proxy/server_test.go
View File

@ -269,6 +269,77 @@ func TestServer_PauseTx(t *testing.T) {
}
}
func TestServer_ModifyTx_corrupt(t *testing.T) {
scheme := "unix"
srcAddr, dstAddr := newUnixAddr(), newUnixAddr()
defer func() {
os.RemoveAll(srcAddr)
os.RemoveAll(dstAddr)
}()
ln := listen(t, scheme, dstAddr, transport.TLSInfo{})
defer ln.Close()
p := NewServer(ServerConfig{
Logger: testLogger,
From: url.URL{Scheme: scheme, Host: srcAddr},
To: url.URL{Scheme: scheme, Host: dstAddr},
})
<-p.Ready()
defer p.Close()
p.ModifyTx(func(d []byte) []byte {
d[len(d)/2]++
return d
})
data := []byte("Hello World!")
send(t, data, scheme, srcAddr, transport.TLSInfo{})
if d := receive(t, ln); bytes.Equal(d, data) {
t.Fatalf("expected corrupted data, got %q", string(d))
}
p.UnmodifyTx()
send(t, data, scheme, srcAddr, transport.TLSInfo{})
if d := receive(t, ln); !bytes.Equal(d, data) {
t.Fatalf("expected uncorrupted data, got %q", string(d))
}
}
func TestServer_ModifyTx_packet_loss(t *testing.T) {
scheme := "unix"
srcAddr, dstAddr := newUnixAddr(), newUnixAddr()
defer func() {
os.RemoveAll(srcAddr)
os.RemoveAll(dstAddr)
}()
ln := listen(t, scheme, dstAddr, transport.TLSInfo{})
defer ln.Close()
p := NewServer(ServerConfig{
Logger: testLogger,
From: url.URL{Scheme: scheme, Host: srcAddr},
To: url.URL{Scheme: scheme, Host: dstAddr},
})
<-p.Ready()
defer p.Close()
// 50% packet loss
p.ModifyTx(func(d []byte) []byte {
half := len(d) / 2
return d[:half:half]
})
data := []byte("Hello World!")
send(t, data, scheme, srcAddr, transport.TLSInfo{})
if d := receive(t, ln); bytes.Equal(d, data) {
t.Fatalf("expected corrupted data, got %q", string(d))
}
p.UnmodifyTx()
send(t, data, scheme, srcAddr, transport.TLSInfo{})
if d := receive(t, ln); !bytes.Equal(d, data) {
t.Fatalf("expected uncorrupted data, got %q", string(d))
}
}
func TestServer_BlackholeTx(t *testing.T) {
scheme := "unix"
srcAddr, dstAddr := newUnixAddr(), newUnixAddr()
@ -319,41 +390,6 @@ func TestServer_BlackholeTx(t *testing.T) {
}
}
func TestServer_CorruptTx(t *testing.T) {
scheme := "unix"
srcAddr, dstAddr := newUnixAddr(), newUnixAddr()
defer func() {
os.RemoveAll(srcAddr)
os.RemoveAll(dstAddr)
}()
ln := listen(t, scheme, dstAddr, transport.TLSInfo{})
defer ln.Close()
p := NewServer(ServerConfig{
Logger: testLogger,
From: url.URL{Scheme: scheme, Host: srcAddr},
To: url.URL{Scheme: scheme, Host: dstAddr},
})
<-p.Ready()
defer p.Close()
p.CorruptTx(func(d []byte) []byte {
d[len(d)/2]++
return d
})
data := []byte("Hello World!")
send(t, data, scheme, srcAddr, transport.TLSInfo{})
if d := receive(t, ln); bytes.Equal(d, data) {
t.Fatalf("expected corrupted data, got %q", string(d))
}
p.UncorruptTx()
send(t, data, scheme, srcAddr, transport.TLSInfo{})
if d := receive(t, ln); !bytes.Equal(d, data) {
t.Fatalf("expected uncorrupted data, got %q", string(d))
}
}
func TestServer_Shutdown(t *testing.T) {
scheme := "unix"
srcAddr, dstAddr := newUnixAddr(), newUnixAddr()
@ -372,8 +408,8 @@ func TestServer_Shutdown(t *testing.T) {
<-p.Ready()
defer p.Close()
px, _ := p.(*proxyServer)
px.listener.Close()
s, _ := p.(*server)
s.listener.Close()
time.Sleep(200 * time.Millisecond)
data := []byte("Hello World!")