onProcess is responsible for executing the onConnect/onRequest function serially, and make sure the connection has been closed correctly if user call c.Close() in onConnect/onRequest function.
(onConnect OnConnect, onRequest OnRequest)
| 177 | // onProcess is responsible for executing the onConnect/onRequest function serially, |
| 178 | // and make sure the connection has been closed correctly if user call c.Close() in onConnect/onRequest function. |
| 179 | func (c *connection) onProcess(onConnect OnConnect, onRequest OnRequest) (processed bool) { |
| 180 | // task already exists |
| 181 | if !c.lock(processing) { |
| 182 | return false |
| 183 | } |
| 184 | |
| 185 | task := func() { |
| 186 | panicked := true |
| 187 | defer func() { |
| 188 | if !panicked { |
| 189 | return |
| 190 | } |
| 191 | // cannot use recover() here, since we don't want to break the panic stack |
| 192 | c.unlock(processing) |
| 193 | if c.IsActive() { |
| 194 | c.Close() |
| 195 | } else { |
| 196 | c.closeCallback(false, false) |
| 197 | } |
| 198 | }() |
| 199 | // trigger onConnect first |
| 200 | if onConnect != nil && c.changeState(connStateNone, connStateConnected) { |
| 201 | c.ctx = onConnect(c.ctx, c) |
| 202 | if !c.IsActive() && c.changeState(connStateConnected, connStateDisconnected) { |
| 203 | // since we hold connecting lock, so we should help to call onDisconnect here |
| 204 | onDisconnect, _ := c.onDisconnectCallback.Load().(OnDisconnect) |
| 205 | if onDisconnect != nil { |
| 206 | onDisconnect(c.ctx, c) |
| 207 | } |
| 208 | } |
| 209 | c.unlock(connecting) |
| 210 | } |
| 211 | START: |
| 212 | // The `onRequest` must be executed at least once if conn have any readable data, |
| 213 | // which is in order to cover the `send & close by peer` case. |
| 214 | if onRequest != nil && c.Reader().Len() > 0 { |
| 215 | _ = onRequest(c.ctx, c) |
| 216 | } |
| 217 | // The processing loop must ensure that the connection meets `IsActive`. |
| 218 | // `onRequest` must either eventually read all the input data or actively Close the connection, |
| 219 | // otherwise the goroutine will fall into a dead loop. |
| 220 | var closedBy who |
| 221 | for { |
| 222 | closedBy = c.status(closing) |
| 223 | // close by user or not processable |
| 224 | if closedBy == user || onRequest == nil || c.Reader().Len() == 0 { |
| 225 | break |
| 226 | } |
| 227 | _ = onRequest(c.ctx, c) |
| 228 | } |
| 229 | // handling callback if connection has been closed. |
| 230 | if closedBy != none { |
| 231 | // if closed by user when processing, it "may" needs detach |
| 232 | needDetach := closedBy == user |
| 233 | // Here is a corner case that operator will be detached twice: |
| 234 | // If server closed the connection(client OnHup will detach op first and closeBy=poller), |
| 235 | // and then client's OnRequest function also closed the connection(closeBy=user). |
| 236 | // But operator already prevent that detach twice will not cause any problem |