22
33import java .util .LinkedHashMap ;
44import java .util .Map ;
5+ import java .util .Objects ;
6+ import java .util .Set ;
57import java .util .UUID ;
8+ import java .util .function .Consumer ;
69
710/**
811 * Builds, encodes and decodes the canonical BabelQueue envelope — the single Java
@@ -21,6 +24,23 @@ public final class EnvelopeCodec {
2124 /** The value stamped into {@code meta.lang} for envelopes produced here. */
2225 public static final String SOURCE_LANG = "java" ;
2326
27+ /** Top-level keys the codec maps to {@link Envelope} components ({@code urn} is the inbound alias of {@code job}). */
28+ private static final Set <String > KNOWN_TOP_LEVEL =
29+ Set .of ("job" , "urn" , "trace_id" , "data" , "meta" , "attempts" , "dead_letter" );
30+
31+ /** {@code meta} keys the codec maps to {@link Meta} components. */
32+ private static final Set <String > KNOWN_META =
33+ Set .of ("id" , "queue" , "lang" , "schema_version" , "created_at" );
34+
35+ /** Forbidden top-level keys (message-envelope.md §10) — never captured, never encoded. */
36+ private static final Set <String > FORBIDDEN_TOP_LEVEL = Set .of ("timestamp" );
37+
38+ /** Forbidden {@code meta} keys (message-envelope.md §10) — never captured, never encoded. */
39+ private static final Set <String > FORBIDDEN_META =
40+ Set .of ("max_retries" , "attempts" , "source" , "ts" );
41+
42+ private static final System .Logger LOGGER = System .getLogger (EnvelopeCodec .class .getName ());
43+
2444 private EnvelopeCodec () {}
2545
2646 /** Build the canonical envelope for a {@code (urn, data)} pair on the default queue. */
@@ -79,7 +99,10 @@ public static Envelope fromMessage(PolyglotMessage message, String queue) {
7999
80100 /**
81101 * Encode the envelope as compact UTF-8 JSON. Slashes and non-ASCII are left
82- * unescaped, matching the other SDK cores; the field order is canonical.
102+ * unescaped, matching the other SDK cores; the field order is canonical. Known
103+ * fields are written first in their canonical order, followed by the unknown keys
104+ * kept in {@link Envelope#extras()} / {@link Meta#extras()} in their original order.
105+ * Forbidden keys (§10) are never written, even if a caller put them into extras.
83106 */
84107 public static String encode (Envelope envelope ) {
85108 Map <String , Object > root = new LinkedHashMap <>();
@@ -95,6 +118,7 @@ public static String encode(Envelope envelope) {
95118 metaMap .put ("lang" , meta .lang ());
96119 metaMap .put ("schema_version" , meta .schemaVersion ());
97120 metaMap .put ("created_at" , meta .createdAt ());
121+ appendExtras (metaMap , meta .extras (), KNOWN_META , FORBIDDEN_META );
98122 }
99123 root .put ("meta" , metaMap );
100124 root .put ("attempts" , envelope .attempts ());
@@ -112,16 +136,46 @@ public static String encode(Envelope envelope) {
112136 root .put ("dead_letter" , dlMap );
113137 }
114138
139+ appendExtras (root , envelope .extras (), KNOWN_TOP_LEVEL , FORBIDDEN_TOP_LEVEL );
115140 return Json .write (root );
116141 }
117142
143+ private static void appendExtras (
144+ Map <String , Object > target , Map <String , Object > extras , Set <String > known , Set <String > forbidden ) {
145+ if (extras == null ) {
146+ return ;
147+ }
148+ for (Map .Entry <String , Object > entry : extras .entrySet ()) {
149+ String key = entry .getKey ();
150+ if (key == null || known .contains (key ) || forbidden .contains (key ) || target .containsKey (key )) {
151+ continue ;
152+ }
153+ target .put (key , entry .getValue ());
154+ }
155+ }
156+
118157 /**
119158 * Parse a raw JSON body into an {@link Envelope}. Malformed or non-object input
120159 * yields an empty envelope (so {@link #accepts} returns {@code false}); the
121160 * {@code urn} inbound alias is resolved into {@code job}. Does not validate the
122161 * contents — call {@link #accepts} first.
162+ *
163+ * <p>Unknown top-level and {@code meta} keys are preserved in the envelope's
164+ * extras. Forbidden keys (message-envelope.md §10) are dropped and a warning is
165+ * logged via {@link System.Logger} at {@code WARNING} level; use
166+ * {@link #decode(String, Consumer)} to receive the warnings directly.
123167 */
124168 public static Envelope decode (String raw ) {
169+ return decode (raw , warning -> LOGGER .log (System .Logger .Level .WARNING , warning ));
170+ }
171+
172+ /**
173+ * Same as {@link #decode(String)}, but each warning — one per forbidden key found
174+ * and dropped, naming its JSON pointer (e.g. {@code /meta/attempts}) — is passed to
175+ * {@code onWarning} instead of the logger. Forbidden keys do not reject the message.
176+ */
177+ public static Envelope decode (String raw , Consumer <String > onWarning ) {
178+ Objects .requireNonNull (onWarning , "onWarning" );
125179 Object parsed ;
126180 try {
127181 parsed = Json .parse (raw );
@@ -140,13 +194,29 @@ public static Envelope decode(String raw) {
140194 }
141195 }
142196
197+ Map <String , Object > extras = new LinkedHashMap <>();
198+ for (Map .Entry <?, ?> entry : map .entrySet ()) {
199+ String key = String .valueOf (entry .getKey ());
200+ if (FORBIDDEN_TOP_LEVEL .contains (key )) {
201+ onWarning .accept (forbiddenWarning ("/" + key ));
202+ } else if (!KNOWN_TOP_LEVEL .contains (key )) {
203+ extras .put (key , entry .getValue ());
204+ }
205+ }
206+
143207 return new Envelope (
144208 job ,
145209 asString (map .get ("trace_id" )),
146210 asMap (map .get ("data" )),
147- parseMeta (map .get ("meta" )),
211+ parseMeta (map .get ("meta" ), onWarning ),
148212 asInt (map .get ("attempts" ), 0 ),
149- parseDeadLetter (map .get ("dead_letter" )));
213+ parseDeadLetter (map .get ("dead_letter" )),
214+ extras );
215+ }
216+
217+ private static String forbiddenWarning (String pointer ) {
218+ return "BabelQueue: dropped forbidden envelope key " + pointer
219+ + " (message-envelope.md §10); it will not be re-emitted" ;
150220 }
151221
152222 /** The message URN — the canonical {@code job}, with the {@code urn} alias resolved by {@link #decode}. */
@@ -176,17 +246,27 @@ private static Envelope empty() {
176246 return new Envelope (null , null , null , null , 0 , null );
177247 }
178248
179- private static Meta parseMeta (Object value ) {
249+ private static Meta parseMeta (Object value , Consumer < String > onWarning ) {
180250 Map <String , Object > map = asMap (value );
181251 if (map == null ) {
182252 return null ;
183253 }
254+ Map <String , Object > extras = new LinkedHashMap <>();
255+ for (Map .Entry <String , Object > entry : map .entrySet ()) {
256+ String key = entry .getKey ();
257+ if (FORBIDDEN_META .contains (key )) {
258+ onWarning .accept (forbiddenWarning ("/meta/" + key ));
259+ } else if (!KNOWN_META .contains (key )) {
260+ extras .put (key , entry .getValue ());
261+ }
262+ }
184263 return new Meta (
185264 asString (map .get ("id" )),
186265 asString (map .get ("queue" )),
187266 asString (map .get ("lang" )),
188267 asInt (map .get ("schema_version" ), 0 ),
189- asLong (map .get ("created_at" ), 0L ));
268+ asLong (map .get ("created_at" ), 0L ),
269+ extras );
190270 }
191271
192272 private static DeadLetter parseDeadLetter (Object value ) {
0 commit comments