frame.go 42 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968
  1. // Copyright (c) 2012 The gocql Authors. All rights reserved.
  2. // Use of this source code is governed by a BSD-style
  3. // license that can be found in the LICENSE file.
  4. package gocql
  5. import (
  6. "context"
  7. "errors"
  8. "fmt"
  9. "io"
  10. "io/ioutil"
  11. "net"
  12. "runtime"
  13. "strings"
  14. "time"
  15. )
  16. type unsetColumn struct{}
  17. // UnsetValue represents a value used in a query binding that will be ignored by Cassandra.
  18. //
  19. // By setting a field to the unset value Cassandra will ignore the write completely.
  20. // The main advantage is the ability to keep the same prepared statement even when you don't
  21. // want to update some fields, where before you needed to make another prepared statement.
  22. //
  23. // UnsetValue is only available when using the version 4 of the protocol.
  24. var UnsetValue = unsetColumn{}
  25. type namedValue struct {
  26. name string
  27. value interface{}
  28. }
  29. // NamedValue produce a value which will bind to the named parameter in a query
  30. func NamedValue(name string, value interface{}) interface{} {
  31. return &namedValue{
  32. name: name,
  33. value: value,
  34. }
  35. }
  36. const (
  37. protoDirectionMask = 0x80
  38. protoVersionMask = 0x7F
  39. protoVersion1 = 0x01
  40. protoVersion2 = 0x02
  41. protoVersion3 = 0x03
  42. protoVersion4 = 0x04
  43. protoVersion5 = 0x05
  44. maxFrameSize = 256 * 1024 * 1024
  45. )
  46. type protoVersion byte
  47. func (p protoVersion) request() bool {
  48. return p&protoDirectionMask == 0x00
  49. }
  50. func (p protoVersion) response() bool {
  51. return p&protoDirectionMask == 0x80
  52. }
  53. func (p protoVersion) version() byte {
  54. return byte(p) & protoVersionMask
  55. }
  56. func (p protoVersion) String() string {
  57. dir := "REQ"
  58. if p.response() {
  59. dir = "RESP"
  60. }
  61. return fmt.Sprintf("[version=%d direction=%s]", p.version(), dir)
  62. }
  63. type frameOp byte
  64. const (
  65. // header ops
  66. opError frameOp = 0x00
  67. opStartup frameOp = 0x01
  68. opReady frameOp = 0x02
  69. opAuthenticate frameOp = 0x03
  70. opOptions frameOp = 0x05
  71. opSupported frameOp = 0x06
  72. opQuery frameOp = 0x07
  73. opResult frameOp = 0x08
  74. opPrepare frameOp = 0x09
  75. opExecute frameOp = 0x0A
  76. opRegister frameOp = 0x0B
  77. opEvent frameOp = 0x0C
  78. opBatch frameOp = 0x0D
  79. opAuthChallenge frameOp = 0x0E
  80. opAuthResponse frameOp = 0x0F
  81. opAuthSuccess frameOp = 0x10
  82. )
  83. func (f frameOp) String() string {
  84. switch f {
  85. case opError:
  86. return "ERROR"
  87. case opStartup:
  88. return "STARTUP"
  89. case opReady:
  90. return "READY"
  91. case opAuthenticate:
  92. return "AUTHENTICATE"
  93. case opOptions:
  94. return "OPTIONS"
  95. case opSupported:
  96. return "SUPPORTED"
  97. case opQuery:
  98. return "QUERY"
  99. case opResult:
  100. return "RESULT"
  101. case opPrepare:
  102. return "PREPARE"
  103. case opExecute:
  104. return "EXECUTE"
  105. case opRegister:
  106. return "REGISTER"
  107. case opEvent:
  108. return "EVENT"
  109. case opBatch:
  110. return "BATCH"
  111. case opAuthChallenge:
  112. return "AUTH_CHALLENGE"
  113. case opAuthResponse:
  114. return "AUTH_RESPONSE"
  115. case opAuthSuccess:
  116. return "AUTH_SUCCESS"
  117. default:
  118. return fmt.Sprintf("UNKNOWN_OP_%d", f)
  119. }
  120. }
  121. const (
  122. // result kind
  123. resultKindVoid = 1
  124. resultKindRows = 2
  125. resultKindKeyspace = 3
  126. resultKindPrepared = 4
  127. resultKindSchemaChanged = 5
  128. // rows flags
  129. flagGlobalTableSpec int = 0x01
  130. flagHasMorePages int = 0x02
  131. flagNoMetaData int = 0x04
  132. // query flags
  133. flagValues byte = 0x01
  134. flagSkipMetaData byte = 0x02
  135. flagPageSize byte = 0x04
  136. flagWithPagingState byte = 0x08
  137. flagWithSerialConsistency byte = 0x10
  138. flagDefaultTimestamp byte = 0x20
  139. flagWithNameValues byte = 0x40
  140. // header flags
  141. flagCompress byte = 0x01
  142. flagTracing byte = 0x02
  143. flagCustomPayload byte = 0x04
  144. flagWarning byte = 0x08
  145. )
  146. type Consistency uint16
  147. const (
  148. Any Consistency = 0x00
  149. One Consistency = 0x01
  150. Two Consistency = 0x02
  151. Three Consistency = 0x03
  152. Quorum Consistency = 0x04
  153. All Consistency = 0x05
  154. LocalQuorum Consistency = 0x06
  155. EachQuorum Consistency = 0x07
  156. LocalOne Consistency = 0x0A
  157. )
  158. func (c Consistency) String() string {
  159. switch c {
  160. case Any:
  161. return "ANY"
  162. case One:
  163. return "ONE"
  164. case Two:
  165. return "TWO"
  166. case Three:
  167. return "THREE"
  168. case Quorum:
  169. return "QUORUM"
  170. case All:
  171. return "ALL"
  172. case LocalQuorum:
  173. return "LOCAL_QUORUM"
  174. case EachQuorum:
  175. return "EACH_QUORUM"
  176. case LocalOne:
  177. return "LOCAL_ONE"
  178. default:
  179. return fmt.Sprintf("UNKNOWN_CONS_0x%x", uint16(c))
  180. }
  181. }
  182. func (c Consistency) MarshalText() (text []byte, err error) {
  183. return []byte(c.String()), nil
  184. }
  185. func (c *Consistency) UnmarshalText(text []byte) error {
  186. switch string(text) {
  187. case "ANY":
  188. *c = Any
  189. case "ONE":
  190. *c = One
  191. case "TWO":
  192. *c = Two
  193. case "THREE":
  194. *c = Three
  195. case "QUORUM":
  196. *c = Quorum
  197. case "ALL":
  198. *c = All
  199. case "LOCAL_QUORUM":
  200. *c = LocalQuorum
  201. case "EACH_QUORUM":
  202. *c = EachQuorum
  203. case "LOCAL_ONE":
  204. *c = LocalOne
  205. default:
  206. return fmt.Errorf("invalid consistency %q", string(text))
  207. }
  208. return nil
  209. }
  210. func ParseConsistency(s string) Consistency {
  211. var c Consistency
  212. if err := c.UnmarshalText([]byte(strings.ToUpper(s))); err != nil {
  213. panic(err)
  214. }
  215. return c
  216. }
  217. // ParseConsistencyWrapper wraps gocql.ParseConsistency to provide an err
  218. // return instead of a panic
  219. func ParseConsistencyWrapper(s string) (consistency Consistency, err error) {
  220. err = consistency.UnmarshalText([]byte(strings.ToUpper(s)))
  221. return
  222. }
  223. // MustParseConsistency is the same as ParseConsistency except it returns
  224. // an error (never). It is kept here since breaking changes are not good.
  225. // DEPRECATED: use ParseConsistency if you want a panic on parse error.
  226. func MustParseConsistency(s string) (Consistency, error) {
  227. c, err := ParseConsistencyWrapper(s)
  228. if err != nil {
  229. panic(err)
  230. }
  231. return c, nil
  232. }
  233. type SerialConsistency uint16
  234. const (
  235. Serial SerialConsistency = 0x08
  236. LocalSerial SerialConsistency = 0x09
  237. )
  238. func (s SerialConsistency) String() string {
  239. switch s {
  240. case Serial:
  241. return "SERIAL"
  242. case LocalSerial:
  243. return "LOCAL_SERIAL"
  244. default:
  245. return fmt.Sprintf("UNKNOWN_SERIAL_CONS_0x%x", uint16(s))
  246. }
  247. }
  248. func (s SerialConsistency) MarshalText() (text []byte, err error) {
  249. return []byte(s.String()), nil
  250. }
  251. func (s *SerialConsistency) UnmarshalText(text []byte) error {
  252. switch string(text) {
  253. case "SERIAL":
  254. *s = Serial
  255. case "LOCAL_SERIAL":
  256. *s = LocalSerial
  257. default:
  258. return fmt.Errorf("invalid consistency %q", string(text))
  259. }
  260. return nil
  261. }
  262. const (
  263. apacheCassandraTypePrefix = "org.apache.cassandra.db.marshal."
  264. )
  265. var (
  266. ErrFrameTooBig = errors.New("frame length is bigger than the maximum allowed")
  267. )
  268. const maxFrameHeaderSize = 9
  269. func writeInt(p []byte, n int32) {
  270. p[0] = byte(n >> 24)
  271. p[1] = byte(n >> 16)
  272. p[2] = byte(n >> 8)
  273. p[3] = byte(n)
  274. }
  275. func readInt(p []byte) int32 {
  276. return int32(p[0])<<24 | int32(p[1])<<16 | int32(p[2])<<8 | int32(p[3])
  277. }
  278. func writeShort(p []byte, n uint16) {
  279. p[0] = byte(n >> 8)
  280. p[1] = byte(n)
  281. }
  282. func readShort(p []byte) uint16 {
  283. return uint16(p[0])<<8 | uint16(p[1])
  284. }
  285. type frameHeader struct {
  286. version protoVersion
  287. flags byte
  288. stream int
  289. op frameOp
  290. length int
  291. customPayload map[string][]byte
  292. warnings []string
  293. }
  294. func (f frameHeader) String() string {
  295. return fmt.Sprintf("[header version=%s flags=0x%x stream=%d op=%s length=%d]", f.version, f.flags, f.stream, f.op, f.length)
  296. }
  297. func (f frameHeader) Header() frameHeader {
  298. return f
  299. }
  300. const defaultBufSize = 128
  301. type ObservedFrameHeader struct {
  302. Version protoVersion
  303. Flags byte
  304. Stream int16
  305. Opcode frameOp
  306. Length int32
  307. // StartHeader is the time we started reading the frame header off the network connection.
  308. Start time.Time
  309. // EndHeader is the time we finished reading the frame header off the network connection.
  310. End time.Time
  311. }
  312. func (f ObservedFrameHeader) String() string {
  313. return fmt.Sprintf("[observed header version=%s flags=0x%x stream=%d op=%s length=%d]", f.Version, f.Flags, f.Stream, f.Opcode, f.Length)
  314. }
  315. // FrameHeaderObserver is the interface implemented by frame observers / stat collectors.
  316. //
  317. // Experimental, this interface and use may change
  318. type FrameHeaderObserver interface {
  319. // ObserveFrameHeader gets called on every received frame header.
  320. ObserveFrameHeader(context.Context, ObservedFrameHeader)
  321. }
  322. // a framer is responsible for reading, writing and parsing frames on a single stream
  323. type framer struct {
  324. r io.Reader
  325. w io.Writer
  326. proto byte
  327. // flags are for outgoing flags, enabling compression and tracing etc
  328. flags byte
  329. compres Compressor
  330. headSize int
  331. // if this frame was read then the header will be here
  332. header *frameHeader
  333. // if tracing flag is set this is not nil
  334. traceID []byte
  335. // holds a ref to the whole byte slice for rbuf so that it can be reset to
  336. // 0 after a read.
  337. readBuffer []byte
  338. rbuf []byte
  339. wbuf []byte
  340. }
  341. func newFramer(r io.Reader, w io.Writer, compressor Compressor, version byte) *framer {
  342. f := &framer{
  343. wbuf: make([]byte, defaultBufSize),
  344. readBuffer: make([]byte, defaultBufSize),
  345. }
  346. var flags byte
  347. if compressor != nil {
  348. flags |= flagCompress
  349. }
  350. version &= protoVersionMask
  351. headSize := 8
  352. if version > protoVersion2 {
  353. headSize = 9
  354. }
  355. f.compres = compressor
  356. f.proto = version
  357. f.flags = flags
  358. f.headSize = headSize
  359. f.r = r
  360. f.rbuf = f.readBuffer[:0]
  361. f.w = w
  362. f.wbuf = f.wbuf[:0]
  363. f.header = nil
  364. f.traceID = nil
  365. return f
  366. }
  367. type frame interface {
  368. Header() frameHeader
  369. }
  370. func readHeader(r io.Reader, p []byte) (head frameHeader, err error) {
  371. _, err = io.ReadFull(r, p[:1])
  372. if err != nil {
  373. return frameHeader{}, err
  374. }
  375. version := p[0] & protoVersionMask
  376. if version < protoVersion1 || version > protoVersion4 {
  377. return frameHeader{}, fmt.Errorf("gocql: unsupported protocol response version: %d", version)
  378. }
  379. headSize := 9
  380. if version < protoVersion3 {
  381. headSize = 8
  382. }
  383. _, err = io.ReadFull(r, p[1:headSize])
  384. if err != nil {
  385. return frameHeader{}, err
  386. }
  387. p = p[:headSize]
  388. head.version = protoVersion(p[0])
  389. head.flags = p[1]
  390. if version > protoVersion2 {
  391. if len(p) != 9 {
  392. return frameHeader{}, fmt.Errorf("not enough bytes to read header require 9 got: %d", len(p))
  393. }
  394. head.stream = int(int16(p[2])<<8 | int16(p[3]))
  395. head.op = frameOp(p[4])
  396. head.length = int(readInt(p[5:]))
  397. } else {
  398. if len(p) != 8 {
  399. return frameHeader{}, fmt.Errorf("not enough bytes to read header require 8 got: %d", len(p))
  400. }
  401. head.stream = int(int8(p[2]))
  402. head.op = frameOp(p[3])
  403. head.length = int(readInt(p[4:]))
  404. }
  405. return head, nil
  406. }
  407. // explicitly enables tracing for the framers outgoing requests
  408. func (f *framer) trace() {
  409. f.flags |= flagTracing
  410. }
  411. // reads a frame form the wire into the framers buffer
  412. func (f *framer) readFrame(head *frameHeader) error {
  413. if head.length < 0 {
  414. return fmt.Errorf("frame body length can not be less than 0: %d", head.length)
  415. } else if head.length > maxFrameSize {
  416. // need to free up the connection to be used again
  417. _, err := io.CopyN(ioutil.Discard, f.r, int64(head.length))
  418. if err != nil {
  419. return fmt.Errorf("error whilst trying to discard frame with invalid length: %v", err)
  420. }
  421. return ErrFrameTooBig
  422. }
  423. if cap(f.readBuffer) >= head.length {
  424. f.rbuf = f.readBuffer[:head.length]
  425. } else {
  426. f.readBuffer = make([]byte, head.length)
  427. f.rbuf = f.readBuffer
  428. }
  429. // assume the underlying reader takes care of timeouts and retries
  430. n, err := io.ReadFull(f.r, f.rbuf)
  431. if err != nil {
  432. return fmt.Errorf("unable to read frame body: read %d/%d bytes: %v", n, head.length, err)
  433. }
  434. if head.flags&flagCompress == flagCompress {
  435. if f.compres == nil {
  436. return NewErrProtocol("no compressor available with compressed frame body")
  437. }
  438. f.rbuf, err = f.compres.Decode(f.rbuf)
  439. if err != nil {
  440. return err
  441. }
  442. }
  443. f.header = head
  444. return nil
  445. }
  446. func (f *framer) parseFrame() (frame frame, err error) {
  447. defer func() {
  448. if r := recover(); r != nil {
  449. if _, ok := r.(runtime.Error); ok {
  450. panic(r)
  451. }
  452. err = r.(error)
  453. }
  454. }()
  455. if f.header.version.request() {
  456. return nil, NewErrProtocol("got a request frame from server: %v", f.header.version)
  457. }
  458. if f.header.flags&flagTracing == flagTracing {
  459. f.readTrace()
  460. }
  461. if f.header.flags&flagWarning == flagWarning {
  462. f.header.warnings = f.readStringList()
  463. }
  464. if f.header.flags&flagCustomPayload == flagCustomPayload {
  465. f.header.customPayload = f.readBytesMap()
  466. }
  467. // assumes that the frame body has been read into rbuf
  468. switch f.header.op {
  469. case opError:
  470. frame = f.parseErrorFrame()
  471. case opReady:
  472. frame = f.parseReadyFrame()
  473. case opResult:
  474. frame, err = f.parseResultFrame()
  475. case opSupported:
  476. frame = f.parseSupportedFrame()
  477. case opAuthenticate:
  478. frame = f.parseAuthenticateFrame()
  479. case opAuthChallenge:
  480. frame = f.parseAuthChallengeFrame()
  481. case opAuthSuccess:
  482. frame = f.parseAuthSuccessFrame()
  483. case opEvent:
  484. frame = f.parseEventFrame()
  485. default:
  486. return nil, NewErrProtocol("unknown op in frame header: %s", f.header.op)
  487. }
  488. return
  489. }
  490. func (f *framer) parseErrorFrame() frame {
  491. code := f.readInt()
  492. msg := f.readString()
  493. errD := errorFrame{
  494. frameHeader: *f.header,
  495. code: code,
  496. message: msg,
  497. }
  498. switch code {
  499. case errUnavailable:
  500. cl := f.readConsistency()
  501. required := f.readInt()
  502. alive := f.readInt()
  503. return &RequestErrUnavailable{
  504. errorFrame: errD,
  505. Consistency: cl,
  506. Required: required,
  507. Alive: alive,
  508. }
  509. case errWriteTimeout:
  510. cl := f.readConsistency()
  511. received := f.readInt()
  512. blockfor := f.readInt()
  513. writeType := f.readString()
  514. return &RequestErrWriteTimeout{
  515. errorFrame: errD,
  516. Consistency: cl,
  517. Received: received,
  518. BlockFor: blockfor,
  519. WriteType: writeType,
  520. }
  521. case errReadTimeout:
  522. cl := f.readConsistency()
  523. received := f.readInt()
  524. blockfor := f.readInt()
  525. dataPresent := f.readByte()
  526. return &RequestErrReadTimeout{
  527. errorFrame: errD,
  528. Consistency: cl,
  529. Received: received,
  530. BlockFor: blockfor,
  531. DataPresent: dataPresent,
  532. }
  533. case errAlreadyExists:
  534. ks := f.readString()
  535. table := f.readString()
  536. return &RequestErrAlreadyExists{
  537. errorFrame: errD,
  538. Keyspace: ks,
  539. Table: table,
  540. }
  541. case errUnprepared:
  542. stmtId := f.readShortBytes()
  543. return &RequestErrUnprepared{
  544. errorFrame: errD,
  545. StatementId: copyBytes(stmtId), // defensively copy
  546. }
  547. case errReadFailure:
  548. res := &RequestErrReadFailure{
  549. errorFrame: errD,
  550. }
  551. res.Consistency = f.readConsistency()
  552. res.Received = f.readInt()
  553. res.BlockFor = f.readInt()
  554. res.DataPresent = f.readByte() != 0
  555. return res
  556. case errWriteFailure:
  557. res := &RequestErrWriteFailure{
  558. errorFrame: errD,
  559. }
  560. res.Consistency = f.readConsistency()
  561. res.Received = f.readInt()
  562. res.BlockFor = f.readInt()
  563. res.NumFailures = f.readInt()
  564. res.WriteType = f.readString()
  565. return res
  566. case errFunctionFailure:
  567. res := &RequestErrFunctionFailure{
  568. errorFrame: errD,
  569. }
  570. res.Keyspace = f.readString()
  571. res.Function = f.readString()
  572. res.ArgTypes = f.readStringList()
  573. return res
  574. case errCDCWriteFailure:
  575. res := &RequestErrCDCWriteFailure{
  576. errorFrame: errD,
  577. }
  578. return res
  579. case errInvalid, errBootstrapping, errConfig, errCredentials, errOverloaded,
  580. errProtocol, errServer, errSyntax, errTruncate, errUnauthorized:
  581. // TODO(zariel): we should have some distinct types for these errors
  582. return errD
  583. default:
  584. panic(fmt.Errorf("unknown error code: 0x%x", errD.code))
  585. }
  586. }
  587. func (f *framer) writeHeader(flags byte, op frameOp, stream int) {
  588. f.wbuf = f.wbuf[:0]
  589. f.wbuf = append(f.wbuf,
  590. f.proto,
  591. flags,
  592. )
  593. if f.proto > protoVersion2 {
  594. f.wbuf = append(f.wbuf,
  595. byte(stream>>8),
  596. byte(stream),
  597. )
  598. } else {
  599. f.wbuf = append(f.wbuf,
  600. byte(stream),
  601. )
  602. }
  603. // pad out length
  604. f.wbuf = append(f.wbuf,
  605. byte(op),
  606. 0,
  607. 0,
  608. 0,
  609. 0,
  610. )
  611. }
  612. func (f *framer) setLength(length int) {
  613. p := 4
  614. if f.proto > protoVersion2 {
  615. p = 5
  616. }
  617. f.wbuf[p+0] = byte(length >> 24)
  618. f.wbuf[p+1] = byte(length >> 16)
  619. f.wbuf[p+2] = byte(length >> 8)
  620. f.wbuf[p+3] = byte(length)
  621. }
  622. func (f *framer) finishWrite() error {
  623. if len(f.wbuf) > maxFrameSize {
  624. // huge app frame, lets remove it so it doesn't bloat the heap
  625. f.wbuf = make([]byte, defaultBufSize)
  626. return ErrFrameTooBig
  627. }
  628. if f.wbuf[1]&flagCompress == flagCompress {
  629. if f.compres == nil {
  630. panic("compress flag set with no compressor")
  631. }
  632. // TODO: only compress frames which are big enough
  633. compressed, err := f.compres.Encode(f.wbuf[f.headSize:])
  634. if err != nil {
  635. return err
  636. }
  637. f.wbuf = append(f.wbuf[:f.headSize], compressed...)
  638. }
  639. length := len(f.wbuf) - f.headSize
  640. f.setLength(length)
  641. _, err := f.w.Write(f.wbuf)
  642. if err != nil {
  643. return err
  644. }
  645. return nil
  646. }
  647. func (f *framer) readTrace() {
  648. f.traceID = f.readUUID().Bytes()
  649. }
  650. type readyFrame struct {
  651. frameHeader
  652. }
  653. func (f *framer) parseReadyFrame() frame {
  654. return &readyFrame{
  655. frameHeader: *f.header,
  656. }
  657. }
  658. type supportedFrame struct {
  659. frameHeader
  660. supported map[string][]string
  661. }
  662. // TODO: if we move the body buffer onto the frameHeader then we only need a single
  663. // framer, and can move the methods onto the header.
  664. func (f *framer) parseSupportedFrame() frame {
  665. return &supportedFrame{
  666. frameHeader: *f.header,
  667. supported: f.readStringMultiMap(),
  668. }
  669. }
  670. type writeStartupFrame struct {
  671. opts map[string]string
  672. }
  673. func (w writeStartupFrame) String() string {
  674. return fmt.Sprintf("[startup opts=%+v]", w.opts)
  675. }
  676. func (w *writeStartupFrame) writeFrame(f *framer, streamID int) error {
  677. f.writeHeader(f.flags&^flagCompress, opStartup, streamID)
  678. f.writeStringMap(w.opts)
  679. return f.finishWrite()
  680. }
  681. type writePrepareFrame struct {
  682. statement string
  683. }
  684. func (w *writePrepareFrame) writeFrame(f *framer, streamID int) error {
  685. f.writeHeader(f.flags, opPrepare, streamID)
  686. f.writeLongString(w.statement)
  687. return f.finishWrite()
  688. }
  689. func (f *framer) readTypeInfo() TypeInfo {
  690. // TODO: factor this out so the same code paths can be used to parse custom
  691. // types and other types, as much of the logic will be duplicated.
  692. id := f.readShort()
  693. simple := NativeType{
  694. proto: f.proto,
  695. typ: Type(id),
  696. }
  697. if simple.typ == TypeCustom {
  698. simple.custom = f.readString()
  699. if cassType := getApacheCassandraType(simple.custom); cassType != TypeCustom {
  700. simple.typ = cassType
  701. }
  702. }
  703. switch simple.typ {
  704. case TypeTuple:
  705. n := f.readShort()
  706. tuple := TupleTypeInfo{
  707. NativeType: simple,
  708. Elems: make([]TypeInfo, n),
  709. }
  710. for i := 0; i < int(n); i++ {
  711. tuple.Elems[i] = f.readTypeInfo()
  712. }
  713. return tuple
  714. case TypeUDT:
  715. udt := UDTTypeInfo{
  716. NativeType: simple,
  717. }
  718. udt.KeySpace = f.readString()
  719. udt.Name = f.readString()
  720. n := f.readShort()
  721. udt.Elements = make([]UDTField, n)
  722. for i := 0; i < int(n); i++ {
  723. field := &udt.Elements[i]
  724. field.Name = f.readString()
  725. field.Type = f.readTypeInfo()
  726. }
  727. return udt
  728. case TypeMap, TypeList, TypeSet:
  729. collection := CollectionType{
  730. NativeType: simple,
  731. }
  732. if simple.typ == TypeMap {
  733. collection.Key = f.readTypeInfo()
  734. }
  735. collection.Elem = f.readTypeInfo()
  736. return collection
  737. }
  738. return simple
  739. }
  740. type preparedMetadata struct {
  741. resultMetadata
  742. // proto v4+
  743. pkeyColumns []int
  744. }
  745. func (r preparedMetadata) String() string {
  746. return fmt.Sprintf("[prepared flags=0x%x pkey=%v paging_state=% X columns=%v col_count=%d actual_col_count=%d]", r.flags, r.pkeyColumns, r.pagingState, r.columns, r.colCount, r.actualColCount)
  747. }
  748. func (f *framer) parsePreparedMetadata() preparedMetadata {
  749. // TODO: deduplicate this from parseMetadata
  750. meta := preparedMetadata{}
  751. meta.flags = f.readInt()
  752. meta.colCount = f.readInt()
  753. if meta.colCount < 0 {
  754. panic(fmt.Errorf("received negative column count: %d", meta.colCount))
  755. }
  756. meta.actualColCount = meta.colCount
  757. if f.proto >= protoVersion4 {
  758. pkeyCount := f.readInt()
  759. pkeys := make([]int, pkeyCount)
  760. for i := 0; i < pkeyCount; i++ {
  761. pkeys[i] = int(f.readShort())
  762. }
  763. meta.pkeyColumns = pkeys
  764. }
  765. if meta.flags&flagHasMorePages == flagHasMorePages {
  766. meta.pagingState = copyBytes(f.readBytes())
  767. }
  768. if meta.flags&flagNoMetaData == flagNoMetaData {
  769. return meta
  770. }
  771. var keyspace, table string
  772. globalSpec := meta.flags&flagGlobalTableSpec == flagGlobalTableSpec
  773. if globalSpec {
  774. keyspace = f.readString()
  775. table = f.readString()
  776. }
  777. var cols []ColumnInfo
  778. if meta.colCount < 1000 {
  779. // preallocate columninfo to avoid excess copying
  780. cols = make([]ColumnInfo, meta.colCount)
  781. for i := 0; i < meta.colCount; i++ {
  782. f.readCol(&cols[i], &meta.resultMetadata, globalSpec, keyspace, table)
  783. }
  784. } else {
  785. // use append, huge number of columns usually indicates a corrupt frame or
  786. // just a huge row.
  787. for i := 0; i < meta.colCount; i++ {
  788. var col ColumnInfo
  789. f.readCol(&col, &meta.resultMetadata, globalSpec, keyspace, table)
  790. cols = append(cols, col)
  791. }
  792. }
  793. meta.columns = cols
  794. return meta
  795. }
  796. type resultMetadata struct {
  797. flags int
  798. // only if flagPageState
  799. pagingState []byte
  800. columns []ColumnInfo
  801. colCount int
  802. // this is a count of the total number of columns which can be scanned,
  803. // it is at minimum len(columns) but may be larger, for instance when a column
  804. // is a UDT or tuple.
  805. actualColCount int
  806. }
  807. func (r resultMetadata) String() string {
  808. return fmt.Sprintf("[metadata flags=0x%x paging_state=% X columns=%v]", r.flags, r.pagingState, r.columns)
  809. }
  810. func (f *framer) readCol(col *ColumnInfo, meta *resultMetadata, globalSpec bool, keyspace, table string) {
  811. if !globalSpec {
  812. col.Keyspace = f.readString()
  813. col.Table = f.readString()
  814. } else {
  815. col.Keyspace = keyspace
  816. col.Table = table
  817. }
  818. col.Name = f.readString()
  819. col.TypeInfo = f.readTypeInfo()
  820. switch v := col.TypeInfo.(type) {
  821. // maybe also UDT
  822. case TupleTypeInfo:
  823. // -1 because we already included the tuple column
  824. meta.actualColCount += len(v.Elems) - 1
  825. }
  826. }
  827. func (f *framer) parseResultMetadata() resultMetadata {
  828. var meta resultMetadata
  829. meta.flags = f.readInt()
  830. meta.colCount = f.readInt()
  831. if meta.colCount < 0 {
  832. panic(fmt.Errorf("received negative column count: %d", meta.colCount))
  833. }
  834. meta.actualColCount = meta.colCount
  835. if meta.flags&flagHasMorePages == flagHasMorePages {
  836. meta.pagingState = copyBytes(f.readBytes())
  837. }
  838. if meta.flags&flagNoMetaData == flagNoMetaData {
  839. return meta
  840. }
  841. var keyspace, table string
  842. globalSpec := meta.flags&flagGlobalTableSpec == flagGlobalTableSpec
  843. if globalSpec {
  844. keyspace = f.readString()
  845. table = f.readString()
  846. }
  847. var cols []ColumnInfo
  848. if meta.colCount < 1000 {
  849. // preallocate columninfo to avoid excess copying
  850. cols = make([]ColumnInfo, meta.colCount)
  851. for i := 0; i < meta.colCount; i++ {
  852. f.readCol(&cols[i], &meta, globalSpec, keyspace, table)
  853. }
  854. } else {
  855. // use append, huge number of columns usually indicates a corrupt frame or
  856. // just a huge row.
  857. for i := 0; i < meta.colCount; i++ {
  858. var col ColumnInfo
  859. f.readCol(&col, &meta, globalSpec, keyspace, table)
  860. cols = append(cols, col)
  861. }
  862. }
  863. meta.columns = cols
  864. return meta
  865. }
  866. type resultVoidFrame struct {
  867. frameHeader
  868. }
  869. func (f *resultVoidFrame) String() string {
  870. return "[result_void]"
  871. }
  872. func (f *framer) parseResultFrame() (frame, error) {
  873. kind := f.readInt()
  874. switch kind {
  875. case resultKindVoid:
  876. return &resultVoidFrame{frameHeader: *f.header}, nil
  877. case resultKindRows:
  878. return f.parseResultRows(), nil
  879. case resultKindKeyspace:
  880. return f.parseResultSetKeyspace(), nil
  881. case resultKindPrepared:
  882. return f.parseResultPrepared(), nil
  883. case resultKindSchemaChanged:
  884. return f.parseResultSchemaChange(), nil
  885. }
  886. return nil, NewErrProtocol("unknown result kind: %x", kind)
  887. }
  888. type resultRowsFrame struct {
  889. frameHeader
  890. meta resultMetadata
  891. // dont parse the rows here as we only need to do it once
  892. numRows int
  893. }
  894. func (f *resultRowsFrame) String() string {
  895. return fmt.Sprintf("[result_rows meta=%v]", f.meta)
  896. }
  897. func (f *framer) parseResultRows() frame {
  898. result := &resultRowsFrame{}
  899. result.meta = f.parseResultMetadata()
  900. result.numRows = f.readInt()
  901. if result.numRows < 0 {
  902. panic(fmt.Errorf("invalid row_count in result frame: %d", result.numRows))
  903. }
  904. return result
  905. }
  906. type resultKeyspaceFrame struct {
  907. frameHeader
  908. keyspace string
  909. }
  910. func (r *resultKeyspaceFrame) String() string {
  911. return fmt.Sprintf("[result_keyspace keyspace=%s]", r.keyspace)
  912. }
  913. func (f *framer) parseResultSetKeyspace() frame {
  914. return &resultKeyspaceFrame{
  915. frameHeader: *f.header,
  916. keyspace: f.readString(),
  917. }
  918. }
  919. type resultPreparedFrame struct {
  920. frameHeader
  921. preparedID []byte
  922. reqMeta preparedMetadata
  923. respMeta resultMetadata
  924. }
  925. func (f *framer) parseResultPrepared() frame {
  926. frame := &resultPreparedFrame{
  927. frameHeader: *f.header,
  928. preparedID: f.readShortBytes(),
  929. reqMeta: f.parsePreparedMetadata(),
  930. }
  931. if f.proto < protoVersion2 {
  932. return frame
  933. }
  934. frame.respMeta = f.parseResultMetadata()
  935. return frame
  936. }
  937. type schemaChangeKeyspace struct {
  938. frameHeader
  939. change string
  940. keyspace string
  941. }
  942. func (f schemaChangeKeyspace) String() string {
  943. return fmt.Sprintf("[event schema_change_keyspace change=%q keyspace=%q]", f.change, f.keyspace)
  944. }
  945. type schemaChangeTable struct {
  946. frameHeader
  947. change string
  948. keyspace string
  949. object string
  950. }
  951. func (f schemaChangeTable) String() string {
  952. return fmt.Sprintf("[event schema_change change=%q keyspace=%q object=%q]", f.change, f.keyspace, f.object)
  953. }
  954. type schemaChangeType struct {
  955. frameHeader
  956. change string
  957. keyspace string
  958. object string
  959. }
  960. type schemaChangeFunction struct {
  961. frameHeader
  962. change string
  963. keyspace string
  964. name string
  965. args []string
  966. }
  967. type schemaChangeAggregate struct {
  968. frameHeader
  969. change string
  970. keyspace string
  971. name string
  972. args []string
  973. }
  974. func (f *framer) parseResultSchemaChange() frame {
  975. if f.proto <= protoVersion2 {
  976. change := f.readString()
  977. keyspace := f.readString()
  978. table := f.readString()
  979. if table != "" {
  980. return &schemaChangeTable{
  981. frameHeader: *f.header,
  982. change: change,
  983. keyspace: keyspace,
  984. object: table,
  985. }
  986. } else {
  987. return &schemaChangeKeyspace{
  988. frameHeader: *f.header,
  989. change: change,
  990. keyspace: keyspace,
  991. }
  992. }
  993. } else {
  994. change := f.readString()
  995. target := f.readString()
  996. // TODO: could just use a separate type for each target
  997. switch target {
  998. case "KEYSPACE":
  999. frame := &schemaChangeKeyspace{
  1000. frameHeader: *f.header,
  1001. change: change,
  1002. }
  1003. frame.keyspace = f.readString()
  1004. return frame
  1005. case "TABLE":
  1006. frame := &schemaChangeTable{
  1007. frameHeader: *f.header,
  1008. change: change,
  1009. }
  1010. frame.keyspace = f.readString()
  1011. frame.object = f.readString()
  1012. return frame
  1013. case "TYPE":
  1014. frame := &schemaChangeType{
  1015. frameHeader: *f.header,
  1016. change: change,
  1017. }
  1018. frame.keyspace = f.readString()
  1019. frame.object = f.readString()
  1020. return frame
  1021. case "FUNCTION":
  1022. frame := &schemaChangeFunction{
  1023. frameHeader: *f.header,
  1024. change: change,
  1025. }
  1026. frame.keyspace = f.readString()
  1027. frame.name = f.readString()
  1028. frame.args = f.readStringList()
  1029. return frame
  1030. case "AGGREGATE":
  1031. frame := &schemaChangeAggregate{
  1032. frameHeader: *f.header,
  1033. change: change,
  1034. }
  1035. frame.keyspace = f.readString()
  1036. frame.name = f.readString()
  1037. frame.args = f.readStringList()
  1038. return frame
  1039. default:
  1040. panic(fmt.Errorf("gocql: unknown SCHEMA_CHANGE target: %q change: %q", target, change))
  1041. }
  1042. }
  1043. }
  1044. type authenticateFrame struct {
  1045. frameHeader
  1046. class string
  1047. }
  1048. func (a *authenticateFrame) String() string {
  1049. return fmt.Sprintf("[authenticate class=%q]", a.class)
  1050. }
  1051. func (f *framer) parseAuthenticateFrame() frame {
  1052. return &authenticateFrame{
  1053. frameHeader: *f.header,
  1054. class: f.readString(),
  1055. }
  1056. }
  1057. type authSuccessFrame struct {
  1058. frameHeader
  1059. data []byte
  1060. }
  1061. func (a *authSuccessFrame) String() string {
  1062. return fmt.Sprintf("[auth_success data=%q]", a.data)
  1063. }
  1064. func (f *framer) parseAuthSuccessFrame() frame {
  1065. return &authSuccessFrame{
  1066. frameHeader: *f.header,
  1067. data: f.readBytes(),
  1068. }
  1069. }
  1070. type authChallengeFrame struct {
  1071. frameHeader
  1072. data []byte
  1073. }
  1074. func (a *authChallengeFrame) String() string {
  1075. return fmt.Sprintf("[auth_challenge data=%q]", a.data)
  1076. }
  1077. func (f *framer) parseAuthChallengeFrame() frame {
  1078. return &authChallengeFrame{
  1079. frameHeader: *f.header,
  1080. data: f.readBytes(),
  1081. }
  1082. }
  1083. type statusChangeEventFrame struct {
  1084. frameHeader
  1085. change string
  1086. host net.IP
  1087. port int
  1088. }
  1089. func (t statusChangeEventFrame) String() string {
  1090. return fmt.Sprintf("[status_change change=%s host=%v port=%v]", t.change, t.host, t.port)
  1091. }
  1092. // essentially the same as statusChange
  1093. type topologyChangeEventFrame struct {
  1094. frameHeader
  1095. change string
  1096. host net.IP
  1097. port int
  1098. }
  1099. func (t topologyChangeEventFrame) String() string {
  1100. return fmt.Sprintf("[topology_change change=%s host=%v port=%v]", t.change, t.host, t.port)
  1101. }
  1102. func (f *framer) parseEventFrame() frame {
  1103. eventType := f.readString()
  1104. switch eventType {
  1105. case "TOPOLOGY_CHANGE":
  1106. frame := &topologyChangeEventFrame{frameHeader: *f.header}
  1107. frame.change = f.readString()
  1108. frame.host, frame.port = f.readInet()
  1109. return frame
  1110. case "STATUS_CHANGE":
  1111. frame := &statusChangeEventFrame{frameHeader: *f.header}
  1112. frame.change = f.readString()
  1113. frame.host, frame.port = f.readInet()
  1114. return frame
  1115. case "SCHEMA_CHANGE":
  1116. // this should work for all versions
  1117. return f.parseResultSchemaChange()
  1118. default:
  1119. panic(fmt.Errorf("gocql: unknown event type: %q", eventType))
  1120. }
  1121. }
  1122. type writeAuthResponseFrame struct {
  1123. data []byte
  1124. }
  1125. func (a *writeAuthResponseFrame) String() string {
  1126. return fmt.Sprintf("[auth_response data=%q]", a.data)
  1127. }
  1128. func (a *writeAuthResponseFrame) writeFrame(framer *framer, streamID int) error {
  1129. return framer.writeAuthResponseFrame(streamID, a.data)
  1130. }
  1131. func (f *framer) writeAuthResponseFrame(streamID int, data []byte) error {
  1132. f.writeHeader(f.flags, opAuthResponse, streamID)
  1133. f.writeBytes(data)
  1134. return f.finishWrite()
  1135. }
  1136. type queryValues struct {
  1137. value []byte
  1138. // optional name, will set With names for values flag
  1139. name string
  1140. isUnset bool
  1141. }
  1142. type queryParams struct {
  1143. consistency Consistency
  1144. // v2+
  1145. skipMeta bool
  1146. values []queryValues
  1147. pageSize int
  1148. pagingState []byte
  1149. serialConsistency SerialConsistency
  1150. // v3+
  1151. defaultTimestamp bool
  1152. defaultTimestampValue int64
  1153. }
  1154. func (q queryParams) String() string {
  1155. return fmt.Sprintf("[query_params consistency=%v skip_meta=%v page_size=%d paging_state=%q serial_consistency=%v default_timestamp=%v values=%v]",
  1156. q.consistency, q.skipMeta, q.pageSize, q.pagingState, q.serialConsistency, q.defaultTimestamp, q.values)
  1157. }
  1158. func (f *framer) writeQueryParams(opts *queryParams) {
  1159. f.writeConsistency(opts.consistency)
  1160. if f.proto == protoVersion1 {
  1161. return
  1162. }
  1163. var flags byte
  1164. if len(opts.values) > 0 {
  1165. flags |= flagValues
  1166. }
  1167. if opts.skipMeta {
  1168. flags |= flagSkipMetaData
  1169. }
  1170. if opts.pageSize > 0 {
  1171. flags |= flagPageSize
  1172. }
  1173. if len(opts.pagingState) > 0 {
  1174. flags |= flagWithPagingState
  1175. }
  1176. if opts.serialConsistency > 0 {
  1177. flags |= flagWithSerialConsistency
  1178. }
  1179. names := false
  1180. // protoV3 specific things
  1181. if f.proto > protoVersion2 {
  1182. if opts.defaultTimestamp {
  1183. flags |= flagDefaultTimestamp
  1184. }
  1185. if len(opts.values) > 0 && opts.values[0].name != "" {
  1186. flags |= flagWithNameValues
  1187. names = true
  1188. }
  1189. }
  1190. f.writeByte(flags)
  1191. if n := len(opts.values); n > 0 {
  1192. f.writeShort(uint16(n))
  1193. for i := 0; i < n; i++ {
  1194. if names {
  1195. f.writeString(opts.values[i].name)
  1196. }
  1197. if opts.values[i].isUnset {
  1198. f.writeUnset()
  1199. } else {
  1200. f.writeBytes(opts.values[i].value)
  1201. }
  1202. }
  1203. }
  1204. if opts.pageSize > 0 {
  1205. f.writeInt(int32(opts.pageSize))
  1206. }
  1207. if len(opts.pagingState) > 0 {
  1208. f.writeBytes(opts.pagingState)
  1209. }
  1210. if opts.serialConsistency > 0 {
  1211. f.writeConsistency(Consistency(opts.serialConsistency))
  1212. }
  1213. if f.proto > protoVersion2 && opts.defaultTimestamp {
  1214. // timestamp in microseconds
  1215. var ts int64
  1216. if opts.defaultTimestampValue != 0 {
  1217. ts = opts.defaultTimestampValue
  1218. } else {
  1219. ts = time.Now().UnixNano() / 1000
  1220. }
  1221. f.writeLong(ts)
  1222. }
  1223. }
  1224. type writeQueryFrame struct {
  1225. statement string
  1226. params queryParams
  1227. }
  1228. func (w *writeQueryFrame) String() string {
  1229. return fmt.Sprintf("[query statement=%q params=%v]", w.statement, w.params)
  1230. }
  1231. func (w *writeQueryFrame) writeFrame(framer *framer, streamID int) error {
  1232. return framer.writeQueryFrame(streamID, w.statement, &w.params)
  1233. }
  1234. func (f *framer) writeQueryFrame(streamID int, statement string, params *queryParams) error {
  1235. f.writeHeader(f.flags, opQuery, streamID)
  1236. f.writeLongString(statement)
  1237. f.writeQueryParams(params)
  1238. return f.finishWrite()
  1239. }
  1240. type frameWriter interface {
  1241. writeFrame(framer *framer, streamID int) error
  1242. }
  1243. type frameWriterFunc func(framer *framer, streamID int) error
  1244. func (f frameWriterFunc) writeFrame(framer *framer, streamID int) error {
  1245. return f(framer, streamID)
  1246. }
  1247. type writeExecuteFrame struct {
  1248. preparedID []byte
  1249. params queryParams
  1250. }
  1251. func (e *writeExecuteFrame) String() string {
  1252. return fmt.Sprintf("[execute id=% X params=%v]", e.preparedID, &e.params)
  1253. }
  1254. func (e *writeExecuteFrame) writeFrame(fr *framer, streamID int) error {
  1255. return fr.writeExecuteFrame(streamID, e.preparedID, &e.params)
  1256. }
  1257. func (f *framer) writeExecuteFrame(streamID int, preparedID []byte, params *queryParams) error {
  1258. f.writeHeader(f.flags, opExecute, streamID)
  1259. f.writeShortBytes(preparedID)
  1260. if f.proto > protoVersion1 {
  1261. f.writeQueryParams(params)
  1262. } else {
  1263. n := len(params.values)
  1264. f.writeShort(uint16(n))
  1265. for i := 0; i < n; i++ {
  1266. if params.values[i].isUnset {
  1267. f.writeUnset()
  1268. } else {
  1269. f.writeBytes(params.values[i].value)
  1270. }
  1271. }
  1272. f.writeConsistency(params.consistency)
  1273. }
  1274. return f.finishWrite()
  1275. }
  1276. // TODO: can we replace BatchStatemt with batchStatement? As they prety much
  1277. // duplicate each other
  1278. type batchStatment struct {
  1279. preparedID []byte
  1280. statement string
  1281. values []queryValues
  1282. }
  1283. type writeBatchFrame struct {
  1284. typ BatchType
  1285. statements []batchStatment
  1286. consistency Consistency
  1287. // v3+
  1288. serialConsistency SerialConsistency
  1289. defaultTimestamp bool
  1290. defaultTimestampValue int64
  1291. }
  1292. func (w *writeBatchFrame) writeFrame(framer *framer, streamID int) error {
  1293. return framer.writeBatchFrame(streamID, w)
  1294. }
  1295. func (f *framer) writeBatchFrame(streamID int, w *writeBatchFrame) error {
  1296. f.writeHeader(f.flags, opBatch, streamID)
  1297. f.writeByte(byte(w.typ))
  1298. n := len(w.statements)
  1299. f.writeShort(uint16(n))
  1300. var flags byte
  1301. for i := 0; i < n; i++ {
  1302. b := &w.statements[i]
  1303. if len(b.preparedID) == 0 {
  1304. f.writeByte(0)
  1305. f.writeLongString(b.statement)
  1306. } else {
  1307. f.writeByte(1)
  1308. f.writeShortBytes(b.preparedID)
  1309. }
  1310. f.writeShort(uint16(len(b.values)))
  1311. for j := range b.values {
  1312. col := b.values[j]
  1313. if f.proto > protoVersion2 && col.name != "" {
  1314. // TODO: move this check into the caller and set a flag on writeBatchFrame
  1315. // to indicate using named values
  1316. if f.proto <= protoVersion5 {
  1317. return fmt.Errorf("gocql: named query values are not supported in batches, please see https://issues.apache.org/jira/browse/CASSANDRA-10246")
  1318. }
  1319. flags |= flagWithNameValues
  1320. f.writeString(col.name)
  1321. }
  1322. if col.isUnset {
  1323. f.writeUnset()
  1324. } else {
  1325. f.writeBytes(col.value)
  1326. }
  1327. }
  1328. }
  1329. f.writeConsistency(w.consistency)
  1330. if f.proto > protoVersion2 {
  1331. if w.serialConsistency > 0 {
  1332. flags |= flagWithSerialConsistency
  1333. }
  1334. if w.defaultTimestamp {
  1335. flags |= flagDefaultTimestamp
  1336. }
  1337. f.writeByte(flags)
  1338. if w.serialConsistency > 0 {
  1339. f.writeConsistency(Consistency(w.serialConsistency))
  1340. }
  1341. if w.defaultTimestamp {
  1342. var ts int64
  1343. if w.defaultTimestampValue != 0 {
  1344. ts = w.defaultTimestampValue
  1345. } else {
  1346. ts = time.Now().UnixNano() / 1000
  1347. }
  1348. f.writeLong(ts)
  1349. }
  1350. }
  1351. return f.finishWrite()
  1352. }
  1353. type writeOptionsFrame struct{}
  1354. func (w *writeOptionsFrame) writeFrame(framer *framer, streamID int) error {
  1355. return framer.writeOptionsFrame(streamID, w)
  1356. }
  1357. func (f *framer) writeOptionsFrame(stream int, _ *writeOptionsFrame) error {
  1358. f.writeHeader(f.flags, opOptions, stream)
  1359. return f.finishWrite()
  1360. }
  1361. type writeRegisterFrame struct {
  1362. events []string
  1363. }
  1364. func (w *writeRegisterFrame) writeFrame(framer *framer, streamID int) error {
  1365. return framer.writeRegisterFrame(streamID, w)
  1366. }
  1367. func (f *framer) writeRegisterFrame(streamID int, w *writeRegisterFrame) error {
  1368. f.writeHeader(f.flags, opRegister, streamID)
  1369. f.writeStringList(w.events)
  1370. return f.finishWrite()
  1371. }
  1372. func (f *framer) readByte() byte {
  1373. if len(f.rbuf) < 1 {
  1374. panic(fmt.Errorf("not enough bytes in buffer to read byte require 1 got: %d", len(f.rbuf)))
  1375. }
  1376. b := f.rbuf[0]
  1377. f.rbuf = f.rbuf[1:]
  1378. return b
  1379. }
  1380. func (f *framer) readInt() (n int) {
  1381. if len(f.rbuf) < 4 {
  1382. panic(fmt.Errorf("not enough bytes in buffer to read int require 4 got: %d", len(f.rbuf)))
  1383. }
  1384. n = int(int32(f.rbuf[0])<<24 | int32(f.rbuf[1])<<16 | int32(f.rbuf[2])<<8 | int32(f.rbuf[3]))
  1385. f.rbuf = f.rbuf[4:]
  1386. return
  1387. }
  1388. func (f *framer) readShort() (n uint16) {
  1389. if len(f.rbuf) < 2 {
  1390. panic(fmt.Errorf("not enough bytes in buffer to read short require 2 got: %d", len(f.rbuf)))
  1391. }
  1392. n = uint16(f.rbuf[0])<<8 | uint16(f.rbuf[1])
  1393. f.rbuf = f.rbuf[2:]
  1394. return
  1395. }
  1396. func (f *framer) readLong() (n int64) {
  1397. if len(f.rbuf) < 8 {
  1398. panic(fmt.Errorf("not enough bytes in buffer to read long require 8 got: %d", len(f.rbuf)))
  1399. }
  1400. n = int64(f.rbuf[0])<<56 | int64(f.rbuf[1])<<48 | int64(f.rbuf[2])<<40 | int64(f.rbuf[3])<<32 |
  1401. int64(f.rbuf[4])<<24 | int64(f.rbuf[5])<<16 | int64(f.rbuf[6])<<8 | int64(f.rbuf[7])
  1402. f.rbuf = f.rbuf[8:]
  1403. return
  1404. }
  1405. func (f *framer) readString() (s string) {
  1406. size := f.readShort()
  1407. if len(f.rbuf) < int(size) {
  1408. panic(fmt.Errorf("not enough bytes in buffer to read string require %d got: %d", size, len(f.rbuf)))
  1409. }
  1410. s = string(f.rbuf[:size])
  1411. f.rbuf = f.rbuf[size:]
  1412. return
  1413. }
  1414. func (f *framer) readLongString() (s string) {
  1415. size := f.readInt()
  1416. if len(f.rbuf) < size {
  1417. panic(fmt.Errorf("not enough bytes in buffer to read long string require %d got: %d", size, len(f.rbuf)))
  1418. }
  1419. s = string(f.rbuf[:size])
  1420. f.rbuf = f.rbuf[size:]
  1421. return
  1422. }
  1423. func (f *framer) readUUID() *UUID {
  1424. if len(f.rbuf) < 16 {
  1425. panic(fmt.Errorf("not enough bytes in buffer to read uuid require %d got: %d", 16, len(f.rbuf)))
  1426. }
  1427. // TODO: how to handle this error, if it is a uuid, then sureley, problems?
  1428. u, _ := UUIDFromBytes(f.rbuf[:16])
  1429. f.rbuf = f.rbuf[16:]
  1430. return &u
  1431. }
  1432. func (f *framer) readStringList() []string {
  1433. size := f.readShort()
  1434. l := make([]string, size)
  1435. for i := 0; i < int(size); i++ {
  1436. l[i] = f.readString()
  1437. }
  1438. return l
  1439. }
  1440. func (f *framer) readBytesInternal() ([]byte, error) {
  1441. size := f.readInt()
  1442. if size < 0 {
  1443. return nil, nil
  1444. }
  1445. if len(f.rbuf) < size {
  1446. return nil, fmt.Errorf("not enough bytes in buffer to read bytes require %d got: %d", size, len(f.rbuf))
  1447. }
  1448. l := f.rbuf[:size]
  1449. f.rbuf = f.rbuf[size:]
  1450. return l, nil
  1451. }
  1452. func (f *framer) readBytes() []byte {
  1453. l, err := f.readBytesInternal()
  1454. if err != nil {
  1455. panic(err)
  1456. }
  1457. return l
  1458. }
  1459. func (f *framer) readShortBytes() []byte {
  1460. size := f.readShort()
  1461. if len(f.rbuf) < int(size) {
  1462. panic(fmt.Errorf("not enough bytes in buffer to read short bytes: require %d got %d", size, len(f.rbuf)))
  1463. }
  1464. l := f.rbuf[:size]
  1465. f.rbuf = f.rbuf[size:]
  1466. return l
  1467. }
  1468. func (f *framer) readInet() (net.IP, int) {
  1469. if len(f.rbuf) < 1 {
  1470. panic(fmt.Errorf("not enough bytes in buffer to read inet size require %d got: %d", 1, len(f.rbuf)))
  1471. }
  1472. size := f.rbuf[0]
  1473. f.rbuf = f.rbuf[1:]
  1474. if !(size == 4 || size == 16) {
  1475. panic(fmt.Errorf("invalid IP size: %d", size))
  1476. }
  1477. if len(f.rbuf) < 1 {
  1478. panic(fmt.Errorf("not enough bytes in buffer to read inet require %d got: %d", size, len(f.rbuf)))
  1479. }
  1480. ip := make([]byte, size)
  1481. copy(ip, f.rbuf[:size])
  1482. f.rbuf = f.rbuf[size:]
  1483. port := f.readInt()
  1484. return net.IP(ip), port
  1485. }
  1486. func (f *framer) readConsistency() Consistency {
  1487. return Consistency(f.readShort())
  1488. }
  1489. func (f *framer) readStringMap() map[string]string {
  1490. size := f.readShort()
  1491. m := make(map[string]string, size)
  1492. for i := 0; i < int(size); i++ {
  1493. k := f.readString()
  1494. v := f.readString()
  1495. m[k] = v
  1496. }
  1497. return m
  1498. }
  1499. func (f *framer) readBytesMap() map[string][]byte {
  1500. size := f.readShort()
  1501. m := make(map[string][]byte, size)
  1502. for i := 0; i < int(size); i++ {
  1503. k := f.readString()
  1504. v := f.readBytes()
  1505. m[k] = v
  1506. }
  1507. return m
  1508. }
  1509. func (f *framer) readStringMultiMap() map[string][]string {
  1510. size := f.readShort()
  1511. m := make(map[string][]string, size)
  1512. for i := 0; i < int(size); i++ {
  1513. k := f.readString()
  1514. v := f.readStringList()
  1515. m[k] = v
  1516. }
  1517. return m
  1518. }
  1519. func (f *framer) writeByte(b byte) {
  1520. f.wbuf = append(f.wbuf, b)
  1521. }
  1522. func appendBytes(p []byte, d []byte) []byte {
  1523. if d == nil {
  1524. return appendInt(p, -1)
  1525. }
  1526. p = appendInt(p, int32(len(d)))
  1527. p = append(p, d...)
  1528. return p
  1529. }
  1530. func appendShort(p []byte, n uint16) []byte {
  1531. return append(p,
  1532. byte(n>>8),
  1533. byte(n),
  1534. )
  1535. }
  1536. func appendInt(p []byte, n int32) []byte {
  1537. return append(p, byte(n>>24),
  1538. byte(n>>16),
  1539. byte(n>>8),
  1540. byte(n))
  1541. }
  1542. func appendLong(p []byte, n int64) []byte {
  1543. return append(p,
  1544. byte(n>>56),
  1545. byte(n>>48),
  1546. byte(n>>40),
  1547. byte(n>>32),
  1548. byte(n>>24),
  1549. byte(n>>16),
  1550. byte(n>>8),
  1551. byte(n),
  1552. )
  1553. }
  1554. // these are protocol level binary types
  1555. func (f *framer) writeInt(n int32) {
  1556. f.wbuf = appendInt(f.wbuf, n)
  1557. }
  1558. func (f *framer) writeShort(n uint16) {
  1559. f.wbuf = appendShort(f.wbuf, n)
  1560. }
  1561. func (f *framer) writeLong(n int64) {
  1562. f.wbuf = appendLong(f.wbuf, n)
  1563. }
  1564. func (f *framer) writeString(s string) {
  1565. f.writeShort(uint16(len(s)))
  1566. f.wbuf = append(f.wbuf, s...)
  1567. }
  1568. func (f *framer) writeLongString(s string) {
  1569. f.writeInt(int32(len(s)))
  1570. f.wbuf = append(f.wbuf, s...)
  1571. }
  1572. func (f *framer) writeUUID(u *UUID) {
  1573. f.wbuf = append(f.wbuf, u[:]...)
  1574. }
  1575. func (f *framer) writeStringList(l []string) {
  1576. f.writeShort(uint16(len(l)))
  1577. for _, s := range l {
  1578. f.writeString(s)
  1579. }
  1580. }
  1581. func (f *framer) writeUnset() {
  1582. // Protocol version 4 specifies that bind variables do not require having a
  1583. // value when executing a statement. Bind variables without a value are
  1584. // called 'unset'. The 'unset' bind variable is serialized as the int
  1585. // value '-2' without following bytes.
  1586. f.writeInt(-2)
  1587. }
  1588. func (f *framer) writeBytes(p []byte) {
  1589. // TODO: handle null case correctly,
  1590. // [bytes] A [int] n, followed by n bytes if n >= 0. If n < 0,
  1591. // no byte should follow and the value represented is `null`.
  1592. if p == nil {
  1593. f.writeInt(-1)
  1594. } else {
  1595. f.writeInt(int32(len(p)))
  1596. f.wbuf = append(f.wbuf, p...)
  1597. }
  1598. }
  1599. func (f *framer) writeShortBytes(p []byte) {
  1600. f.writeShort(uint16(len(p)))
  1601. f.wbuf = append(f.wbuf, p...)
  1602. }
  1603. func (f *framer) writeInet(ip net.IP, port int) {
  1604. f.wbuf = append(f.wbuf,
  1605. byte(len(ip)),
  1606. )
  1607. f.wbuf = append(f.wbuf,
  1608. []byte(ip)...,
  1609. )
  1610. f.writeInt(int32(port))
  1611. }
  1612. func (f *framer) writeConsistency(cons Consistency) {
  1613. f.writeShort(uint16(cons))
  1614. }
  1615. func (f *framer) writeStringMap(m map[string]string) {
  1616. f.writeShort(uint16(len(m)))
  1617. for k, v := range m {
  1618. f.writeString(k)
  1619. f.writeString(v)
  1620. }
  1621. }