Skip to content

Commit a4c84d3

Browse files
DerekBumDerekBum
committed
Different response interfaces for different requests
1 parent 66c136a commit a4c84d3

18 files changed

+697
-208
lines changed

‎connection.go

Lines changed: 17 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -97,9 +97,9 @@ func (d defaultLogger) Report(event ConnLogKind, conn *Connection, v ...interfac
9797
log.Printf("tarantool: last reconnect to %s failed: %s, giving it up",
9898
conn.Addr(), err)
9999
case LogUnexpectedResultId:
100-
respHeader := v[0].(header)
100+
respHeader := v[0].(Header)
101101
log.Printf("tarantool: connection %s got unexpected resultId (%d) in response",
102-
conn.Addr(), respHeader.requestId)
102+
conn.Addr(), respHeader.RequestId)
103103
case LogWatchEventReadFailed:
104104
err := v[0].(error)
105105
log.Printf("tarantool: unable to parse watch event: %s", err)
@@ -818,10 +818,10 @@ func (conn *Connection) reader(r io.Reader, c Conn) {
818818
return
819819
}
820820

821-
resp := &ConnResponse{header: respHeader, buf: buf}
821+
baseResp := &BaseResponse{header: respHeader, buf: buf}
822822
var fut *Future = nil
823-
if iproto.Type(respHeader.code) == iproto.IPROTO_EVENT {
824-
if event, err := readWatchEvent(&resp.buf); err == nil {
823+
if iproto.Type(respHeader.Code) == iproto.IPROTO_EVENT {
824+
if event, err := readWatchEvent(&baseResp.buf); err == nil {
825825
events <- event
826826
} else {
827827
err = ClientError{
@@ -831,12 +831,16 @@ func (conn *Connection) reader(r io.Reader, c Conn) {
831831
conn.opts.Logger.Report(LogWatchEventReadFailed, conn, err)
832832
}
833833
continue
834-
} else if respHeader.code == PushCode {
835-
if fut = conn.peekFuture(respHeader.requestId); fut != nil {
834+
} else if respHeader.Code == PushCode {
835+
if fut = conn.peekFuture(respHeader.RequestId); fut != nil {
836+
resp := fut.req.CreateResponse(respHeader)
837+
resp.SetBuf(buf)
836838
fut.AppendPush(resp)
837839
}
838840
} else {
839-
if fut = conn.fetchFuture(respHeader.requestId); fut != nil {
841+
if fut = conn.fetchFuture(respHeader.RequestId); fut != nil {
842+
resp := fut.req.CreateResponse(respHeader)
843+
resp.SetBuf(buf)
840844
fut.SetResponse(resp)
841845
conn.markDone(fut)
842846
}
@@ -873,8 +877,10 @@ func (conn *Connection) eventer(events <-chan connWatchEvent) {
873877
}
874878
}
875879

876-
func (conn *Connection) newFuture(ctx context.Context) (fut *Future) {
880+
func (conn *Connection) newFuture(req Request) (fut *Future) {
881+
ctx := req.Ctx()
877882
fut = NewFuture()
883+
fut.SetRequest(req)
878884
if conn.rlimit != nil && conn.opts.RLimitAction == RLimitDrop {
879885
select {
880886
case conn.rlimit <- struct{}{}:
@@ -984,7 +990,7 @@ func (conn *Connection) decrementRequestCnt() {
984990
func (conn *Connection) send(req Request, streamId uint64) *Future {
985991
conn.incrementRequestCnt()
986992

987-
fut := conn.newFuture(req.Ctx())
993+
fut := conn.newFuture(req)
988994
if fut.ready == nil {
989995
conn.decrementRequestCnt()
990996
return fut
@@ -1053,7 +1059,7 @@ func (conn *Connection) putFuture(fut *Future, req Request, streamId uint64) {
10531059

10541060
if req.Async() {
10551061
if fut = conn.fetchFuture(reqid); fut != nil {
1056-
resp := &ConnResponse{}
1062+
resp := req.CreateResponse(Header{})
10571063
fut.SetResponse(resp)
10581064
conn.markDone(fut)
10591065
}

‎crud/common.go

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -84,6 +84,13 @@ func (req baseRequest) Async() bool {
8484
return req.impl.Async()
8585
}
8686

87+
// CreateResponse creates a response for the baseRequest.
88+
func (req baseRequest) CreateResponse(header tarantool.Header) tarantool.Response {
89+
resp := tarantool.BaseResponse{}
90+
resp.SetHeader(header)
91+
return &resp
92+
}
93+
8794
type spaceRequest struct {
8895
baseRequest
8996
space string

‎crud/select.go

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -133,3 +133,10 @@ func (req SelectRequest) Context(ctx context.Context) SelectRequest {
133133

134134
return req
135135
}
136+
137+
// CreateResponse creates a response for the SelectRequest.
138+
func (req SelectRequest) CreateResponse(header tarantool.Header) tarantool.Response {
139+
resp := tarantool.SelectResponse{}
140+
resp.SetHeader(header)
141+
return &resp
142+
}

‎dial.go

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -511,12 +511,12 @@ func readResponse(r io.Reader) (Response, error) {
511511

512512
respBytes, err := read(r, lenbuf[:])
513513
if err != nil {
514-
return &ConnResponse{}, fmt.Errorf("read error: %w", err)
514+
return &BaseResponse{}, fmt.Errorf("read error: %w", err)
515515
}
516516

517517
buf := smallBuf{b: respBytes}
518518
respHeader, err := decodeHeader(msgpack.NewDecoder(&smallBuf{}), &buf)
519-
resp := &ConnResponse{header: respHeader, buf: buf}
519+
resp := &BaseResponse{header: respHeader, buf: buf}
520520
if err != nil {
521521
return resp, fmt.Errorf("decode response header error: %w", err)
522522
}

‎example_test.go

Lines changed: 49 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -570,8 +570,13 @@ func ExampleExecuteRequest() {
570570
data, err := resp.Decode()
571571
fmt.Println("Error", err)
572572
fmt.Println("Data", data)
573-
fmt.Println("MetaData", resp.MetaData())
574-
fmt.Println("SQL Info", resp.SQLInfo())
573+
exResp, ok := resp.(*tarantool.ExecuteResponse)
574+
if !ok {
575+
fmt.Printf("wrong response type")
576+
return
577+
}
578+
fmt.Println("MetaData", exResp.MetaData())
579+
fmt.Println("SQL Info", exResp.SQLInfo())
575580

576581
// There are 4 options to pass named parameters to an SQL query:
577582
// 1) The simple map;
@@ -608,8 +613,13 @@ func ExampleExecuteRequest() {
608613
data, err = resp.Decode()
609614
fmt.Println("Error", err)
610615
fmt.Println("Data", data)
611-
fmt.Println("MetaData", resp.MetaData())
612-
fmt.Println("SQL Info", resp.SQLInfo())
616+
exResp, ok = resp.(*tarantool.ExecuteResponse)
617+
if !ok {
618+
fmt.Printf("wrong response type")
619+
return
620+
}
621+
fmt.Println("MetaData", exResp.MetaData())
622+
fmt.Println("SQL Info", exResp.SQLInfo())
613623

614624
// 2)
615625
req = req.Args(sqlBind2)
@@ -619,8 +629,13 @@ func ExampleExecuteRequest() {
619629
data, err = resp.Decode()
620630
fmt.Println("Error", err)
621631
fmt.Println("Data", data)
622-
fmt.Println("MetaData", resp.MetaData())
623-
fmt.Println("SQL Info", resp.SQLInfo())
632+
exResp, ok = resp.(*tarantool.ExecuteResponse)
633+
if !ok {
634+
fmt.Printf("wrong response type")
635+
return
636+
}
637+
fmt.Println("MetaData", exResp.MetaData())
638+
fmt.Println("SQL Info", exResp.SQLInfo())
624639

625640
// 3)
626641
req = req.Args(sqlBind3)
@@ -630,8 +645,13 @@ func ExampleExecuteRequest() {
630645
data, err = resp.Decode()
631646
fmt.Println("Error", err)
632647
fmt.Println("Data", data)
633-
fmt.Println("MetaData", resp.MetaData())
634-
fmt.Println("SQL Info", resp.SQLInfo())
648+
exResp, ok = resp.(*tarantool.ExecuteResponse)
649+
if !ok {
650+
fmt.Printf("wrong response type")
651+
return
652+
}
653+
fmt.Println("MetaData", exResp.MetaData())
654+
fmt.Println("SQL Info", exResp.SQLInfo())
635655

636656
// 4)
637657
req = req.Args(sqlBind4)
@@ -641,8 +661,13 @@ func ExampleExecuteRequest() {
641661
data, err = resp.Decode()
642662
fmt.Println("Error", err)
643663
fmt.Println("Data", data)
644-
fmt.Println("MetaData", resp.MetaData())
645-
fmt.Println("SQL Info", resp.SQLInfo())
664+
exResp, ok = resp.(*tarantool.ExecuteResponse)
665+
if !ok {
666+
fmt.Printf("wrong response type")
667+
return
668+
}
669+
fmt.Println("MetaData", exResp.MetaData())
670+
fmt.Println("SQL Info", exResp.SQLInfo())
646671

647672
// The way to pass positional arguments to an SQL query.
648673
req = tarantool.NewExecuteRequest(
@@ -654,8 +679,13 @@ func ExampleExecuteRequest() {
654679
data, err = resp.Decode()
655680
fmt.Println("Error", err)
656681
fmt.Println("Data", data)
657-
fmt.Println("MetaData", resp.MetaData())
658-
fmt.Println("SQL Info", resp.SQLInfo())
682+
exResp, ok = resp.(*tarantool.ExecuteResponse)
683+
if !ok {
684+
fmt.Printf("wrong response type")
685+
return
686+
}
687+
fmt.Println("MetaData", exResp.MetaData())
688+
fmt.Println("SQL Info", exResp.SQLInfo())
659689

660690
// The way to pass SQL expression with using custom packing/unpacking for
661691
// a type.
@@ -680,8 +710,13 @@ func ExampleExecuteRequest() {
680710
data, err = resp.Decode()
681711
fmt.Println("Error", err)
682712
fmt.Println("Data", data)
683-
fmt.Println("MetaData", resp.MetaData())
684-
fmt.Println("SQL Info", resp.SQLInfo())
713+
exResp, ok = resp.(*tarantool.ExecuteResponse)
714+
if !ok {
715+
fmt.Printf("wrong response type")
716+
return
717+
}
718+
fmt.Println("MetaData", exResp.MetaData())
719+
fmt.Println("SQL Info", exResp.SQLInfo())
685720
}
686721

687722
func getTestTxnDialer() tarantool.Dialer {

‎future.go

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ import (
88
// Future is a handle for asynchronous request.
99
type Future struct {
1010
requestId uint32
11+
req Request
1112
next *Future
1213
timeout time.Duration
1314
mutex sync.Mutex
@@ -150,6 +151,11 @@ func (fut *Future) AppendPush(resp Response) {
150151
fut.ready <- struct{}{}
151152
}
152153

154+
// SetRequest sets a request, for which the future was created.
155+
func (fut *Future) SetRequest(req Request) {
156+
fut.req = req
157+
}
158+
153159
// SetResponse sets a response for the future and finishes the future.
154160
func (fut *Future) SetResponse(resp Response) {
155161
fut.mutex.Lock()

‎future_test.go

Lines changed: 10 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -48,7 +48,7 @@ func TestFutureGetIteratorNoItems(t *testing.T) {
4848
}
4949

5050
func TestFutureGetIteratorNoResponse(t *testing.T) {
51-
push := &ConnResponse{}
51+
push := &BaseResponse{}
5252
fut := NewFuture()
5353
fut.AppendPush(push)
5454

@@ -64,7 +64,7 @@ func TestFutureGetIteratorNoResponse(t *testing.T) {
6464
}
6565

6666
func TestFutureGetIteratorNoResponseTimeout(t *testing.T) {
67-
push := &ConnResponse{}
67+
push := &BaseResponse{}
6868
fut := NewFuture()
6969
fut.AppendPush(push)
7070

@@ -80,8 +80,8 @@ func TestFutureGetIteratorNoResponseTimeout(t *testing.T) {
8080
}
8181

8282
func TestFutureGetIteratorResponseOnTimeout(t *testing.T) {
83-
push := &ConnResponse{}
84-
resp := &ConnResponse{}
83+
push := &BaseResponse{}
84+
resp := &BaseResponse{}
8585
fut := NewFuture()
8686
fut.AppendPush(push)
8787

@@ -119,8 +119,8 @@ func TestFutureGetIteratorResponseOnTimeout(t *testing.T) {
119119
}
120120

121121
func TestFutureGetIteratorFirstResponse(t *testing.T) {
122-
resp1 := &ConnResponse{}
123-
resp2 := &ConnResponse{}
122+
resp1 := &BaseResponse{}
123+
resp2 := &BaseResponse{}
124124
fut := NewFuture()
125125
fut.SetResponse(resp1)
126126
fut.SetResponse(resp2)
@@ -155,7 +155,7 @@ func TestFutureGetIteratorFirstError(t *testing.T) {
155155
}
156156

157157
func TestFutureGetIteratorResponse(t *testing.T) {
158-
responses := []*ConnResponse{
158+
responses := []*BaseResponse{
159159
{},
160160
{},
161161
{},
@@ -189,7 +189,7 @@ func TestFutureGetIteratorResponse(t *testing.T) {
189189

190190
func TestFutureGetIteratorError(t *testing.T) {
191191
const errMsg = "error message"
192-
responses := []*ConnResponse{
192+
responses := []*BaseResponse{
193193
{},
194194
{},
195195
}
@@ -226,14 +226,14 @@ func TestFutureGetIteratorError(t *testing.T) {
226226

227227
func TestFutureSetStateRaceCondition(t *testing.T) {
228228
err := errors.New("any error")
229-
resp := &ConnResponse{}
229+
resp := &BaseResponse{}
230230

231231
for i := 0; i < 1000; i++ {
232232
fut := NewFuture()
233233
for j := 0; j < 9; j++ {
234234
go func(opt int) {
235235
if opt%3 == 0 {
236-
respAppend := &ConnResponse{}
236+
respAppend := &BaseResponse{}
237237
fut.AppendPush(respAppend)
238238
} else if opt%3 == 1 {
239239
fut.SetError(err)

‎pool/connection_pool_test.go

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2548,7 +2548,11 @@ func TestNewPrepared(t *testing.T) {
25482548
if reflect.DeepEqual(data[0], []interface{}{1, "test"}) {
25492549
t.Error("Select with named arguments failed")
25502550
}
2551-
metaData := resp.MetaData()
2551+
prepResp, ok := resp.(*tarantool.PrepareResponse)
2552+
if !ok {
2553+
t.Fatalf("Not a Prepare response")
2554+
}
2555+
metaData := prepResp.MetaData()
25522556
if metaData[0].FieldType != "unsigned" ||
25532557
metaData[0].FieldName != "NAME0" ||
25542558
metaData[1].FieldType != "string" ||

‎pool/connector_test.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -135,7 +135,7 @@ type baseRequestMock struct {
135135
mode Mode
136136
}
137137

138-
var reqResp tarantool.Response = &tarantool.ConnResponse{}
138+
var reqResp tarantool.Response = &tarantool.BaseResponse{}
139139
var errReq error = errors.New("response error")
140140
var reqFuture *tarantool.Future = &tarantool.Future{}
141141

‎prepared.go

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -95,6 +95,13 @@ func (req *PrepareRequest) Context(ctx context.Context) *PrepareRequest {
9595
return req
9696
}
9797

98+
// CreateResponse creates a response for the PrepareRequest.
99+
func (req *PrepareRequest) CreateResponse(header Header) Response {
100+
resp := PrepareResponse{}
101+
resp.SetHeader(header)
102+
return &resp
103+
}
104+
98105
// UnprepareRequest helps you to create an unprepare request object for
99106
// execution by a Connection.
100107
type UnprepareRequest struct {
@@ -175,3 +182,10 @@ func (req *ExecutePreparedRequest) Context(ctx context.Context) *ExecutePrepared
175182
req.ctx = ctx
176183
return req
177184
}
185+
186+
// CreateResponse creates a response for the ExecutePreparedRequest.
187+
func (req *ExecutePreparedRequest) CreateResponse(header Header) Response {
188+
resp := PrepareResponse{}
189+
resp.SetHeader(header)
190+
return &resp
191+
}

0 commit comments

Comments
 (0)