| 1218 | } |
| 1219 | |
| 1220 | func (a *csAttempt) finish(err error) { |
| 1221 | a.mu.Lock() |
| 1222 | if a.finished { |
| 1223 | a.mu.Unlock() |
| 1224 | return |
| 1225 | } |
| 1226 | a.finished = true |
| 1227 | if err == io.EOF { |
| 1228 | // Ending a stream with EOF indicates a success. |
| 1229 | err = nil |
| 1230 | } |
| 1231 | var tr metadata.MD |
| 1232 | if a.transportStream != nil { |
| 1233 | a.transportStream.Close(err) |
| 1234 | tr = a.transportStream.Trailer() |
| 1235 | } |
| 1236 | |
| 1237 | if a.pickResult.Done != nil { |
| 1238 | br := false |
| 1239 | if a.transportStream != nil { |
| 1240 | br = a.transportStream.BytesReceived() |
| 1241 | } |
| 1242 | a.pickResult.Done(balancer.DoneInfo{ |
| 1243 | Err: err, |
| 1244 | Trailer: tr, |
| 1245 | BytesSent: a.transportStream != nil, |
| 1246 | BytesReceived: br, |
| 1247 | ServerLoad: balancerload.Parse(tr), |
| 1248 | }) |
| 1249 | } |
| 1250 | if a.statsHandler != nil { |
| 1251 | a.statsHandler.HandleRPC(a.ctx, &stats.End{ |
| 1252 | Client: true, |
| 1253 | BeginTime: a.beginTime, |
| 1254 | EndTime: time.Now(), |
| 1255 | Trailer: tr, |
| 1256 | Error: err, |
| 1257 | }) |
| 1258 | } |
| 1259 | if a.trInfo != nil && a.trInfo.tr != nil { |
| 1260 | if err == nil { |
| 1261 | a.trInfo.tr.LazyPrintf("RPC: [OK]") |
| 1262 | } else { |
| 1263 | a.trInfo.tr.LazyPrintf("RPC: [%v]", err) |
| 1264 | a.trInfo.tr.SetError() |
| 1265 | } |
| 1266 | a.trInfo.tr.Finish() |
| 1267 | a.trInfo.tr = nil |
| 1268 | } |
| 1269 | a.mu.Unlock() |
| 1270 | } |
| 1271 | |
| 1272 | // newNonRetryClientStream creates a ClientStream with the specified transport, on the |
| 1273 | // given addrConn. |