createPassthrough creates the go routines that will read from and write to the connections.
(args passthroughArguments)
| 182 | |
| 183 | // createPassthrough creates the go routines that will read from and write to the connections. |
| 184 | func createPassthrough(args passthroughArguments) { |
| 185 | lastQuery := "" |
| 186 | writeMutex := &sync.Mutex{} |
| 187 | go func() { |
| 188 | defer args.terminate.Done() |
| 189 | for { |
| 190 | psqlMessage, err := args.psqlConnBackend.Receive() |
| 191 | if err != nil { |
| 192 | errStr := err.Error() |
| 193 | if errStr != "unexpected EOF" && !strings.HasSuffix(errStr, "use of closed network connection") { |
| 194 | fmt.Println(err) |
| 195 | } |
| 196 | return |
| 197 | } |
| 198 | switch msg := psqlMessage.(type) { |
| 199 | case *pgproto3.Query: |
| 200 | writeMutex.Lock() |
| 201 | if len(lastQuery) == 0 { |
| 202 | lastQuery = msg.String |
| 203 | } |
| 204 | writeMutex.Unlock() |
| 205 | for _, query := range args.skippedQueries { |
| 206 | if strings.HasPrefix(msg.String, query) { |
| 207 | // An empty query allows for the proper response messages to be sent. |
| 208 | msg.String = ";" |
| 209 | break |
| 210 | } |
| 211 | } |
| 212 | for _, query := range args.breakpointQueries { |
| 213 | if strings.HasPrefix(msg.String, query) { |
| 214 | args.triggerBreakpoint(msg.String) |
| 215 | break |
| 216 | } |
| 217 | } |
| 218 | case *pgproto3.Terminate: |
| 219 | return |
| 220 | } |
| 221 | args.doltgresConnFrontend.Send(psqlMessage) |
| 222 | if err = args.doltgresConnFrontend.Flush(); err != nil { |
| 223 | errStr := err.Error() |
| 224 | if errStr != "unexpected EOF" && !strings.HasSuffix(errStr, "use of closed network connection") { |
| 225 | fmt.Println(err) |
| 226 | } |
| 227 | return |
| 228 | } |
| 229 | } |
| 230 | }() |
| 231 | go func() { |
| 232 | for { |
| 233 | doltgresMessage, err := args.doltgresConnFrontend.Receive() |
| 234 | if err != nil { |
| 235 | errStr := err.Error() |
| 236 | if errStr != "unexpected EOF" && |
| 237 | !strings.HasSuffix(errStr, "use of closed network connection") && |
| 238 | !strings.HasSuffix(errStr, "An existing connection was forcibly closed by the remote host.") { |
| 239 | fmt.Println(err) |
| 240 | } |
| 241 | return |
no test coverage detected