worryg0d commented on code in PR #1822:
URL:
https://github.com/apache/cassandra-gocql-driver/pull/1822#discussion_r1934288925
##########
conn.go:
##########
@@ -777,6 +774,172 @@ func (c *Conn) handleTimeout() {
}
}
+func (c *Conn) recvSegment(ctx context.Context) error {
+ var (
+ frame []byte
+ isSelfContained bool
+ err error
+ )
+
+ // Read frame based on compression
+ if c.compressor != nil {
+ frame, isSelfContained, err = readCompressedSegment(c.r,
c.compressor)
+ } else {
+ frame, isSelfContained, err = readUncompressedSegment(c.r)
+ }
+ if err != nil {
+ return err
+ }
+
+ if isSelfContained {
+ return c.processAllFramesInSegment(ctx, bytes.NewReader(frame))
+ }
+
+ head, err := readHeader(bytes.NewReader(frame), c.headerBuf[:])
+ if err != nil {
+ return err
+ }
+
+ const frameHeaderLength = 9
+ buf := bytes.NewBuffer(make([]byte, 0, head.length+frameHeaderLength))
+ buf.Write(frame)
+
+ // Computing how many bytes of message left to read
+ bytesToRead := head.length - len(frame) + frameHeaderLength
+
+ err = c.recvPartialFrames(buf, bytesToRead)
+ if err != nil {
+ return err
+ }
+
+ return c.processFrame(ctx, buf)
+}
+
+// recvPartialFrames reads proto v5 segments from Conn.r and writes decoded
partial frames to dst.
+// It reads data until the bytesToRead is reached.
+// If Conn.compressor is not nil, it processes Compressed Format segments.
+func (c *Conn) recvPartialFrames(dst *bytes.Buffer, bytesToRead int) error {
+ var (
+ read int
+ frame []byte
+ isSelfContained bool
+ err error
+ )
+
+ for read != bytesToRead {
+ // Read frame based on compression
+ if c.compressor != nil {
+ frame, isSelfContained, err =
readCompressedSegment(c.r, c.compressor)
+ } else {
+ frame, isSelfContained, err =
readUncompressedSegment(c.r)
+ }
+ if err != nil {
+ return fmt.Errorf("gocql: failed to read non
self-contained frame: %w", err)
+ }
+
+ if isSelfContained {
+ return fmt.Errorf("gocql: received self-contained
segment, but expected not")
+ }
+
+ if totalLength := dst.Len() + len(frame); totalLength >
dst.Cap() {
+ return fmt.Errorf("gocql: expected partial frame of
length %d, got %d", dst.Cap(), totalLength)
+ }
+
+ // Write the frame to the destination writer
+ n, _ := dst.Write(frame)
+ read += n
+ }
+
+ return nil
+}
+
+func (c *Conn) processAllFramesInSegment(ctx context.Context, r *bytes.Reader)
error {
+ var err error
+ for r.Len() > 0 && err == nil {
+ err = c.processFrame(ctx, r)
+ }
+
+ return err
+}
+
+// ConnReader is like net.Conn but also allows to set timeout duration.
+type ConnReader interface {
+ net.Conn
+
+ // SetTimeout sets timeout duration for reading data form conn
+ SetTimeout(timeout time.Duration)
+
+ // GetTimeout returns timeout duration
+ GetTimeout() time.Duration
+}
+
+// connReader implements ConnReader.
+// It retries to read data up to 5 times or returns error.
Review Comment:
Here I focused on extracting this
[code](https://github.com/apache/cassandra-gocql-driver/blob/8bb171c2edfc696f96dc99c8db3499742d69db13/conn.go#L329)
to the specific `connReader` type to avoid unintentional reading from
`Conn.conn` or `Conn.r`. I didn't aim to change this behavior, because this is
a bit out of the scope of this PR. However, I already added a `connReader`...
Probably changing this behavior should be handled here as well
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
