Skip to content

Commit 018ac21

Browse files
authored
Merge pull request #801 from ngmr/refactor/fix-post-unmarshal
refactor/fix post unmarshal
2 parents e1884c6 + 97f2476 commit 018ac21

39 files changed

Lines changed: 1063 additions & 861 deletions

testify-iiop/src/main/java/testify/iiop/annotation/OrbSteward.java

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,10 @@
2828
import org.omg.PortableServer.POAHelper;
2929
import org.omg.PortableServer.POAManagerPackage.AdapterInactive;
3030
import testify.iiop.annotation.ConfigureOrb.UseWithOrb;
31+
32+
import static org.apache.yoko.util.ThreadLocalStack.CMSF_THREAD_LOCAL;
33+
import static org.apache.yoko.util.ThreadLocalStack.ROFL_THREAD_LOCAL;
34+
import static org.apache.yoko.util.ThreadLocalStack.YASF_THREAD_LOCAL;
3135
import testify.iiop.annotation.ConfigureOrb.UseWithOrb.InitializerScope;
3236

3337
import java.util.Objects;
@@ -107,9 +111,9 @@ public synchronized void close() {
107111
// When ORB shuts down, in-flight requests may not complete their full interceptor lifecycle,
108112
// leaving ThreadLocal state on the stack. Reset to ensure clean state for next test.
109113
// TODO: Remove this workaround once issue #783 is properly fixed in Yoko core.
110-
org.apache.yoko.util.cmsf.CmsfThreadLocal.reset();
111-
org.apache.yoko.util.rofl.RoflThreadLocal.reset();
112-
org.apache.yoko.util.yasf.YasfThreadLocal.reset();
114+
CMSF_THREAD_LOCAL.reset();
115+
ROFL_THREAD_LOCAL.reset();
116+
YASF_THREAD_LOCAL.reset();
113117
}
114118

115119
private boolean isOrbModifier(Class<?> c) {

yoko-core/src/main/java/org/apache/yoko/orb/OB/DowncallStub.java

Lines changed: 5 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -546,7 +546,7 @@ public void AMIRouterPostMarshal(GIOPOutgoingMessage outgoing, OutputStreamHolde
546546
//
547547
// Invoke a request from a portable stub
548548
//
549-
public YokoInputStream invoke(
549+
public org.omg.CORBA.portable.InputStream invoke(
550550
org.omg.CORBA.Object self,
551551
YokoOutputStream out)
552552
throws ApplicationException,
@@ -610,13 +610,11 @@ public YokoInputStream invoke(
610610
throw new ApplicationException(id, in);
611611
} else {
612612
//
613-
// We're using portable stubs, so we'll never
614-
// know the unmarshalled results. Therefore,
615-
// we might as well invoke the interceptors now.
613+
// Wrap the InputStream to trigger interceptors after the first read.
614+
// This ensures the result is available to interceptors and unmarshalling
615+
// exceptions are properly reported via receive_exception.
616616
//
617-
down.postUnmarshal();
618-
619-
return in;
617+
return new WrappedReplyInputStream(in, down);
620618
}
621619
} else {
622620
down.preUnmarshal();

yoko-core/src/main/java/org/apache/yoko/orb/OB/PIManager.java

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -219,6 +219,12 @@ void serverContextSwitch(ServerRequestInfo info) {
219219
impl._OB_contextSwitch();
220220
}
221221

222+
// Call post_marshal on extended interceptors after response marshalling
223+
void serverPostMarshal(ServerRequestInfo info) {
224+
ServerRequestInfo_impl impl = (ServerRequestInfo_impl) info;
225+
impl._OB_postMarshal();
226+
}
227+
222228
// Set the parameter information (SSI case)
223229
void serverParameterDesc(ServerRequestInfo info, ParameterDesc[] argDesc, ParameterDesc retDesc, TypeCode[] exceptionTC) {
224230
ServerRequestInfo_impl impl = (ServerRequestInfo_impl) info;

yoko-core/src/main/java/org/apache/yoko/orb/OB/PIUpcall.java

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -82,6 +82,11 @@ public YokoOutputStream preMarshal() throws LocationForward {
8282
return super.preMarshal();
8383
}
8484

85+
public void postMarshal() {
86+
piManager_.serverPostMarshal(requestInfo_);
87+
super.postMarshal();
88+
}
89+
8590
public void setUserException(Any any) {
8691
try {
8792
UnknownUserException uex = new UnknownUserException(any);

yoko-core/src/main/java/org/apache/yoko/orb/OB/Upcall.java

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -24,10 +24,9 @@
2424
import org.apache.yoko.orb.OCI.GiopVersion;
2525
import org.apache.yoko.orb.OCI.ProfileInfo;
2626
import org.apache.yoko.orb.OCI.TransportInfo;
27+
import org.apache.yoko.io.SimplyCloseable;
2728
import org.apache.yoko.util.Assert;
2829
import org.apache.yoko.util.Timeout;
29-
import org.apache.yoko.util.cmsf.CmsfThreadLocal;
30-
import org.apache.yoko.util.cmsf.CmsfThreadLocal.CmsfOverride;
3130
import org.omg.CORBA.Any;
3231
import org.omg.CORBA.Policy;
3332
import org.omg.CORBA.PolicyManager;
@@ -48,6 +47,7 @@
4847
import static org.apache.yoko.io.Buffer.createWriteBuffer;
4948
import static org.apache.yoko.util.Arrays.EMPTY_INTS;
5049
import static org.apache.yoko.orb.OB.SendingContextRuntimes.SENDING_CONTEXT_RUNTIME;
50+
import static org.apache.yoko.util.ThreadLocalStack.CMSF_THREAD_LOCAL;
5151
import static org.apache.yoko.orb.OCI.GiopVersion.GIOP1_2;
5252

5353
public class Upcall {
@@ -292,7 +292,7 @@ public void setSystemException(SystemException ex) {
292292

293293
private static void createUnknownExceptionServiceContexts(UnknownException ex, ServiceContexts replyContexts) {
294294
final Throwable t = ex.originalEx;
295-
try (CmsfOverride o = CmsfThreadLocal.override()) {
295+
try (SimplyCloseable o = CMSF_THREAD_LOCAL.overrideForInterceptors()) {
296296
CodecPair codecs = CodecPair.getDefaultCodecs(GIOP1_2);
297297
try (YokoOutputStream os = new YokoOutputStream(codecs, GIOP1_2)) {
298298
os._OB_writeEndian();
Lines changed: 174 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,174 @@
1+
/*
2+
* Copyright 2026 IBM Corporation and others.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*
16+
* SPDX-License-Identifier: Apache-2.0
17+
*/
18+
package org.apache.yoko.orb.OB;
19+
20+
import org.apache.yoko.orb.CORBA.YokoInputStream;
21+
import org.omg.CORBA.Any;
22+
import org.omg.CORBA.Context;
23+
import org.omg.CORBA.INTERNAL;
24+
import org.omg.CORBA.ORB;
25+
import org.omg.CORBA.Principal;
26+
import org.omg.CORBA.SystemException;
27+
import org.omg.CORBA.TypeCode;
28+
import org.omg.CORBA.portable.BoxedValueHelper;
29+
import org.omg.CORBA_2_3.portable.InputStream;
30+
31+
import java.io.Serializable;
32+
import java.math.BigDecimal;
33+
import java.util.function.Supplier;
34+
35+
import static org.apache.yoko.util.Exceptions.as;
36+
37+
/**
38+
* Wrapper for InputStream that intercepts the first read operation to trigger
39+
* portable interceptor receive_reply/receive_exception interception points.
40+
* <p>
41+
* This ensures interceptors are called AFTER the stub has unmarshalled at least
42+
* the first value, making the result available to interceptors and ensuring
43+
* unmarshalling exceptions are properly reported via receive_exception.
44+
*/
45+
final class WrappedReplyInputStream extends InputStream {
46+
private final YokoInputStream yokoStream;
47+
private final Downcall downcall;
48+
private boolean readYet;
49+
50+
WrappedReplyInputStream(YokoInputStream yokoStream, Downcall downcall) {
51+
this.yokoStream = yokoStream;
52+
this.downcall = downcall;
53+
}
54+
55+
/**
56+
* Wraps a read operation with first-read interception logic.
57+
* Handles both value-returning and void operations uniformly.
58+
*/
59+
private <T> T wrapRead(Supplier<T> readOp) {
60+
try {
61+
T result = readOp.get();
62+
triggerInterceptorsOnFirstRead();
63+
return result;
64+
} catch (SystemException ex) {
65+
return handleFirstReadException(ex);
66+
}
67+
}
68+
69+
/**
70+
* Wraps a void read operation (array reads) with first-read interception logic.
71+
*/
72+
private void wrapVoidRead(Runnable readOp) {
73+
try {
74+
readOp.run();
75+
triggerInterceptorsOnFirstRead();
76+
} catch (SystemException ex) {
77+
handleFirstReadException(ex);
78+
}
79+
}
80+
81+
/**
82+
* Called before the first read operation returns to trigger interceptors.
83+
* Subsequent reads bypass this check.
84+
*/
85+
private void triggerInterceptorsOnFirstRead() throws SystemException {
86+
if (readYet) return;
87+
readYet = true;
88+
89+
try {
90+
downcall.postUnmarshal();
91+
} catch (LocationForward ex) {
92+
throw as(INTERNAL::new, ex, "Location forward during postUnmarshal");
93+
} catch (FailureException ex) {
94+
throw as(INTERNAL::new, ex, "Failure during postUnmarshal");
95+
}
96+
}
97+
98+
/**
99+
* Handles SystemException during first read by reporting to interceptors
100+
* via receive_exception before propagating the exception.
101+
*/
102+
private <T> T handleFirstReadException(SystemException ex) throws SystemException {
103+
if (readYet) throw ex;
104+
readYet = true;
105+
try {
106+
downcall.unmarshalEx(ex);
107+
downcall.postUnmarshal();
108+
throw ex;
109+
} catch (LocationForward e) {
110+
throw as(INTERNAL::new, e, "Location forward during exception handling");
111+
} catch (FailureException ignored) {
112+
// FailureException is expected when interceptors detect issues during exception handling
113+
// Let it propagate naturally - it will be converted to appropriate CORBA exception
114+
throw ex;
115+
}
116+
}
117+
118+
// Primitive read methods
119+
@Override public boolean read_boolean() { return wrapRead(yokoStream::read_boolean); }
120+
@Override public char read_char() { return wrapRead(yokoStream::read_char); }
121+
@Override public char read_wchar() { return wrapRead(yokoStream::read_wchar); }
122+
@Override public byte read_octet() { return wrapRead(yokoStream::read_octet); }
123+
@Override public short read_short() { return wrapRead(yokoStream::read_short); }
124+
@Override public short read_ushort() { return wrapRead(yokoStream::read_ushort); }
125+
@Override public int read_long() { return wrapRead(yokoStream::read_long); }
126+
@Override public int read_ulong() { return wrapRead(yokoStream::read_ulong); }
127+
@Override public long read_longlong() { return wrapRead(yokoStream::read_longlong); }
128+
@Override public long read_ulonglong() { return wrapRead(yokoStream::read_ulonglong); }
129+
@Override public float read_float() { return wrapRead(yokoStream::read_float); }
130+
@Override public double read_double() { return wrapRead(yokoStream::read_double); }
131+
@Override public String read_string() { return wrapRead(yokoStream::read_string); }
132+
@Override public String read_wstring() { return wrapRead(yokoStream::read_wstring); }
133+
134+
// Array read methods
135+
@Override public void read_boolean_array(boolean[] v, int o, int l) { wrapVoidRead(() -> yokoStream.read_boolean_array(v, o, l)); }
136+
@Override public void read_char_array(char[] v, int o, int l) { wrapVoidRead(() -> yokoStream.read_char_array(v, o, l)); }
137+
@Override public void read_wchar_array(char[] v, int o, int l) { wrapVoidRead(() -> yokoStream.read_wchar_array(v, o, l)); }
138+
@Override public void read_octet_array(byte[] v, int o, int l) { wrapVoidRead(() -> yokoStream.read_octet_array(v, o, l)); }
139+
@Override public void read_short_array(short[] v, int o, int l) { wrapVoidRead(() -> yokoStream.read_short_array(v, o, l)); }
140+
@Override public void read_ushort_array(short[] v, int o, int l) { wrapVoidRead(() -> yokoStream.read_ushort_array(v, o, l)); }
141+
@Override public void read_long_array(int[] v, int o, int l) { wrapVoidRead(() -> yokoStream.read_long_array(v, o, l)); }
142+
@Override public void read_ulong_array(int[] v, int o, int l) { wrapVoidRead(() -> yokoStream.read_ulong_array(v, o, l)); }
143+
@Override public void read_longlong_array(long[] v, int o, int l) { wrapVoidRead(() -> yokoStream.read_longlong_array(v, o, l)); }
144+
@Override public void read_ulonglong_array(long[] v, int o, int l) { wrapVoidRead(() -> yokoStream.read_ulonglong_array(v, o, l)); }
145+
@Override public void read_float_array(float[] v, int o, int l) { wrapVoidRead(() -> yokoStream.read_float_array(v, o, l)); }
146+
@Override public void read_double_array(double[] v, int o, int l) { wrapVoidRead(() -> yokoStream.read_double_array(v, o, l)); }
147+
148+
// Complex type read methods
149+
@Override public org.omg.CORBA.Object read_Object() { return wrapRead(yokoStream::read_Object); }
150+
@Override public TypeCode read_TypeCode() { return wrapRead(yokoStream::read_TypeCode); }
151+
@Override public Any read_any() { return wrapRead(yokoStream::read_any); }
152+
@Override @Deprecated public Principal read_Principal() { return wrapRead(yokoStream::read_Principal); }
153+
@Override public BigDecimal read_fixed() { return wrapRead(yokoStream::read_fixed); }
154+
@SuppressWarnings("rawtypes")
155+
@Override public org.omg.CORBA.Object read_Object(Class clz) { return wrapRead(() -> yokoStream.read_Object(clz)); }
156+
@Override public Context read_Context() { return wrapRead(yokoStream::read_Context); }
157+
158+
// CORBA 2.3 methods
159+
@Override public Serializable read_value() { return wrapRead(yokoStream::read_value); }
160+
@SuppressWarnings("rawtypes")
161+
@Override public Serializable read_value(Class clz) { return wrapRead(() -> yokoStream.read_value(clz)); }
162+
@Override public Serializable read_value(BoxedValueHelper factory) { return wrapRead(() -> yokoStream.read_value(factory)); }
163+
@Override public Serializable read_value(String rep_id) { return wrapRead(() -> yokoStream.read_value(rep_id)); }
164+
@Override public Serializable read_value(Serializable value) { return wrapRead(() -> yokoStream.read_value(value)); }
165+
@Override public Object read_abstract_interface() { return wrapRead(yokoStream::read_abstract_interface); }
166+
@SuppressWarnings("rawtypes")
167+
@Override public Object read_abstract_interface(Class clz) { return wrapRead(() -> yokoStream.read_abstract_interface(clz)); }
168+
169+
// java.io.InputStream method - wrapped for transparency
170+
@Override public int read() { return wrapRead(yokoStream::read); }
171+
172+
// Passthrough method (no interception needed - doesn't read data)
173+
@Override public ORB orb() { return yokoStream.orb(); }
174+
}

yoko-core/src/main/java/org/apache/yoko/orb/PortableInterceptor/ClientRequestInfo_impl.java

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -24,13 +24,10 @@
2424
import org.apache.yoko.orb.OB.PIDowncall;
2525
import org.apache.yoko.orb.OB.Util;
2626
import org.apache.yoko.orb.OCI.ProfileInfo;
27+
import org.apache.yoko.io.SimplyCloseable;
2728
import org.apache.yoko.util.Assert;
2829
import org.apache.yoko.util.CollectionExtras;
2930
import org.apache.yoko.util.Exceptions;
30-
import org.apache.yoko.util.cmsf.CmsfThreadLocal;
31-
import org.apache.yoko.util.cmsf.CmsfThreadLocal.CmsfOverride;
32-
import org.apache.yoko.util.yasf.YasfThreadLocal;
33-
import org.apache.yoko.util.yasf.YasfThreadLocal.YasfOverride;
3431
import org.omg.CORBA.Any;
3532
import org.omg.CORBA.BAD_INV_ORDER;
3633
import org.omg.CORBA.BAD_PARAM;
@@ -49,6 +46,9 @@
4946
import org.omg.PortableInterceptor.ClientRequestInfo;
5047
import org.omg.PortableInterceptor.ClientRequestInterceptor;
5148
import org.omg.PortableInterceptor.ForwardRequest;
49+
50+
import static org.apache.yoko.util.ThreadLocalStack.CMSF_THREAD_LOCAL;
51+
import static org.apache.yoko.util.ThreadLocalStack.YASF_THREAD_LOCAL;
5252
import org.omg.PortableInterceptor.LOCATION_FORWARD;
5353
import org.omg.PortableInterceptor.SUCCESSFUL;
5454
import org.omg.PortableInterceptor.SYSTEM_EXCEPTION;
@@ -269,8 +269,8 @@ public void _OB_request(List<ClientRequestInterceptor> interceptors) throws Loca
269269
argStrategy.setArgsAvail(true);
270270
argStrategy.setExceptAvail(true);
271271

272-
try (CmsfOverride ignored = CmsfThreadLocal.override();
273-
YasfOverride ignored1 = YasfThreadLocal.override()) {
272+
try (SimplyCloseable ignored = CMSF_THREAD_LOCAL.overrideForInterceptors();
273+
SimplyCloseable ignored1 = YASF_THREAD_LOCAL.overrideForInterceptors()) {
274274
for(ClientRequestInterceptor interceptor: interceptors) {
275275
try {
276276
interceptor.send_request(this);
@@ -303,8 +303,8 @@ public void _OB_reply() throws LocationForward {
303303
piCurrent._OB_pushSlotData(newThreadScopePICurrentSlotData);
304304
}
305305

306-
try (CmsfOverride ignored = CmsfThreadLocal.override();
307-
YasfOverride ignored1 = YasfThreadLocal.override()) {
306+
try (SimplyCloseable ignored = CMSF_THREAD_LOCAL.overrideForInterceptors();
307+
SimplyCloseable ignored1 = YASF_THREAD_LOCAL.overrideForInterceptors()) {
308308
for (ClientRequestInterceptor i: CollectionExtras.removeInReverse(interceptors)) {
309309
try {
310310
switch (replyStatus) {
Lines changed: 70 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,70 @@
1+
/*
2+
* Copyright 2026 IBM Corporation and others.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*
16+
* SPDX-License-Identifier: Apache-2.0
17+
*/
18+
package org.apache.yoko.orb.PortableInterceptor;
19+
20+
import org.omg.PortableInterceptor.ForwardRequest;
21+
import org.omg.PortableInterceptor.ServerRequestInfo;
22+
import org.omg.PortableInterceptor.ServerRequestInterceptor;
23+
24+
/**
25+
* Extended server request interceptor interface that provides additional
26+
* interception points for setting up thread-local state after a context switch
27+
* and cleaning up after marshalling.
28+
* This is needed when the request processing switches to a different thread
29+
* between receive_request_service_contexts and argument deserialization,
30+
* and when thread-local state needs to be cleaned up after response marshalling.
31+
*
32+
* All methods have default empty implementations to simplify implementation.
33+
*/
34+
public interface ExtendedServerRequestInterceptor extends ServerRequestInterceptor {
35+
/**
36+
* Called after a context switch to set up thread-local state on the new thread.
37+
* This is invoked before argument deserialization begins, allowing interceptors
38+
* to retrieve data from PICurrent slots and push it onto thread locals.
39+
*
40+
* @param ri the server request info
41+
*/
42+
default void pre_unmarshal(ServerRequestInfo ri) {}
43+
44+
/**
45+
* Called after response marshalling is complete to clean up thread-local state.
46+
* This is invoked after the response has been marshalled, allowing interceptors
47+
* to clean up any thread-local state that was set up during request processing.
48+
*
49+
* @param ri the server request info
50+
*/
51+
default void post_marshal(ServerRequestInfo ri) {}
52+
53+
@Override
54+
default void receive_request_service_contexts(ServerRequestInfo ri) throws ForwardRequest {}
55+
56+
@Override
57+
default void receive_request(ServerRequestInfo ri) throws ForwardRequest {}
58+
59+
@Override
60+
default void send_reply(ServerRequestInfo ri) {}
61+
62+
@Override
63+
default void send_exception(ServerRequestInfo ri) throws ForwardRequest {}
64+
65+
@Override
66+
default void send_other(ServerRequestInfo ri) throws ForwardRequest {}
67+
68+
@Override
69+
default void destroy() {}
70+
}

0 commit comments

Comments
 (0)