Implement server to client ChannelDone

This commit is contained in:
Elara 2022-05-01 11:33:55 -07:00
parent 073ebbd8ba
commit 888d5902ae
3 changed files with 46 additions and 16 deletions

View File

@ -111,6 +111,18 @@ func (c *Client) Call(rcvr, method string, arg interface{}, ret interface{}) err
chElemType := retVal.Type().Elem() chElemType := retVal.Type().Elem()
// For every value received from channel // For every value received from channel
for val := range c.chs[chID] { for val := range c.chs[chID] {
if val.ChannelDone {
// Close and delete channel
c.chMtx.Lock()
close(c.chs[chID])
delete(c.chs, chID)
c.chMtx.Unlock()
// Close return channel
retVal.Close()
break
}
// Get reflect value from channel response // Get reflect value from channel response
rVal := reflect.ValueOf(val.Return) rVal := reflect.ValueOf(val.Return)
@ -124,24 +136,24 @@ func (c *Client) Call(rcvr, method string, arg interface{}, ret interface{}) err
rVal = newVal rVal = newVal
} }
// Try to read from the channel // Send value to channel
recvVal, ok := retVal.TryRecv() retVal.Send(rVal)
// IF the channel cannot be read but the value is valid, }
// the channel must be closed }()
if !ok && recvVal.IsValid() {
// Send done signal go func() {
c.Call("lrpc", "ChannelDone", idStr, nil) for {
val, ok := retVal.Recv()
if !ok && val.IsValid() {
break
}
}
c.Call("lrpc", "ChannelDone", id, nil)
// Close and delete channel // Close and delete channel
c.chMtx.Lock() c.chMtx.Lock()
close(c.chs[chID]) close(c.chs[chID])
delete(c.chs, chID) delete(c.chs, chID)
c.chMtx.Unlock() c.chMtx.Unlock()
break
}
// Send value to channel
retVal.Send(rVal)
}
}() }()
} else { } else {
// IF return value is not a pointer, return error // IF return value is not a pointer, return error

View File

@ -11,6 +11,7 @@ type Request struct {
// Response represents a response returned by the server // Response represents a response returned by the server
type Response struct { type Response struct {
ID string ID string
ChannelDone bool
IsChannel bool IsChannel bool
IsError bool IsError bool
Error string Error string

View File

@ -2,6 +2,7 @@ package server
import ( import (
"errors" "errors"
"io"
"net" "net"
"reflect" "reflect"
"sync" "sync"
@ -243,7 +244,9 @@ func (s *Server) handleConn(c codec.Codec) {
var call types.Request var call types.Request
// Read request using codec // Read request using codec
err := c.Decode(&call) err := c.Decode(&call)
if err != nil { if err == io.EOF {
break
} else if err != nil {
s.sendErr(c, call, nil, err) s.sendErr(c, call, nil, err)
continue continue
} }
@ -285,6 +288,18 @@ func (s *Server) handleConn(c codec.Codec) {
Return: val, Return: val,
}) })
} }
// Cancel context
ctx.Cancel()
// Delete context from map
s.contextsMtx.Lock()
delete(s.contexts, ctx.channelID)
s.contextsMtx.Unlock()
c.Encode(types.Response{
ID: ctx.channelID,
ChannelDone: true,
})
}() }()
} }
@ -321,5 +336,7 @@ func (l lrpc) ChannelDone(_ *Context, id string) {
// Cancel context // Cancel context
ctx.Cancel() ctx.Cancel()
// Delete context from map // Delete context from map
l.srv.contextsMtx.Lock()
delete(l.srv.contexts, id) delete(l.srv.contexts, id)
l.srv.contextsMtx.Unlock()
} }