handleRequest handles one image-processing request. A non-nil error from handleRequest will end request processing.
(ctx context.Context)
| 60 | // handleRequest handles one image-processing request. |
| 61 | // A non-nil error from handleRequest will end request processing. |
| 62 | func (p *processor) handleRequest(ctx context.Context) error { |
| 63 | msg, err := p.requestSub.Receive(ctx) |
| 64 | if err != nil { |
| 65 | // If we can't receive messages, we should stop processing. |
| 66 | return err |
| 67 | } |
| 68 | |
| 69 | var req OrderRequest |
| 70 | if err := json.Unmarshal(msg.Body, &req); err != nil { |
| 71 | // We can't unmarshal the message body. That could be due to a bug or |
| 72 | // change in the frontend, or maybe some other program is sending |
| 73 | // malformed messages. |
| 74 | |
| 75 | // Ack the message, because if we can't unmarshal it then no one else can either. |
| 76 | msg.Ack() |
| 77 | // Don't terminate processing; maybe this is just one bad message. |
| 78 | log.Printf("unmarshaling request: %v", err) |
| 79 | return nil |
| 80 | } |
| 81 | log.Printf("received %+v", req) |
| 82 | order, err := createOrFindOrder(ctx, p.coll, &req) |
| 83 | if err != nil { |
| 84 | // There was a problem with the database, perhaps due to the network. |
| 85 | // Nack the message; perhaps another processor can succeed. |
| 86 | if msg.Nackable() { |
| 87 | msg.Nack() |
| 88 | } |
| 89 | // Assume the database error is permanent: terminate processing. |
| 90 | return err |
| 91 | } |
| 92 | if order == nil { |
| 93 | log.Printf("duplicate finished order %v", req.ID) |
| 94 | // We've already processed this order, so ack the message. |
| 95 | msg.Ack() |
| 96 | return nil |
| 97 | } |
| 98 | // At this point, order is an unfinished order in the database. |
| 99 | // Process it. |
| 100 | err = p.processOrder(ctx, order) |
| 101 | // Any processing errors are saved as notes in the order. |
| 102 | if err != nil { |
| 103 | order.Note = fmt.Sprintf("processing failed: %v", err) |
| 104 | order.OutImage = "" |
| 105 | } |
| 106 | // Save the finished order to the database. |
| 107 | err = p.coll.Update(ctx, order, docstore.Mods{ |
| 108 | "OutImage": order.OutImage, |
| 109 | "Note": order.Note, |
| 110 | "FinishTime": time.Now(), |
| 111 | }) |
| 112 | if err != nil { |
| 113 | // We couldn't save the order to the database. |
| 114 | // Nack the message; perhaps another processor can succeed. |
| 115 | if msg.Nackable() { |
| 116 | msg.Nack() |
| 117 | } |
| 118 | // Assume the database error is permanent: terminate processing. |
| 119 | return err |