Enqueue implements blobcache.QueueAPI.Enqueue
(ctx context.Context, qh blobcache.Handle, msgs []blobcache.Message)
| 41 | |
| 42 | // Enqueue implements blobcache.QueueAPI.Enqueue |
| 43 | func (sys *System) Enqueue(ctx context.Context, qh blobcache.Handle, msgs []blobcache.Message) (*blobcache.InsertResp, error) { |
| 44 | logctx.Debug(ctx, "begin", zap.String("method", "Enqueue"), zap.Stringer("oid", qh.OID)) |
| 45 | defer logctx.Debug(ctx, "done", zap.String("method", "Enqueue"), zap.Stringer("oid", qh.OID)) |
| 46 | q, _, err := sys.resolveQueue(qh, blobcache.Action_QUEUE_ENQUEUE) |
| 47 | if err != nil { |
| 48 | return nil, err |
| 49 | } |
| 50 | maxBytes := q.info.Config.MaxBytesPerMessage |
| 51 | maxHandles := q.info.Config.MaxHandlesPerMessage |
| 52 | for i, msg := range msgs { |
| 53 | if uint32(len(msg.Bytes)) > maxBytes { |
| 54 | return nil, fmt.Errorf("message %d exceeds max bytes per message: %d", i, maxBytes) |
| 55 | } |
| 56 | if uint32(len(msg.Handles)) > maxHandles { |
| 57 | return nil, fmt.Errorf("message %d exceeds max handles per message: %d", i, maxHandles) |
| 58 | } |
| 59 | } |
| 60 | n, err := q.backend.Enqueue(ctx, msgs) |
| 61 | if err != nil { |
| 62 | return nil, err |
| 63 | } |
| 64 | return &blobcache.InsertResp{Success: uint32(n)}, nil |
| 65 | } |
| 66 | |
| 67 | var _ blobcache.QueueAPI = &System{} |
nothing calls this directly
no test coverage detected