001: /**
002: * Licensed to the Apache Software Foundation (ASF) under one
003: * or more contributor license agreements. See the NOTICE file
004: * distributed with this work for additional information
005: * regarding copyright ownership. The ASF licenses this file
006: * to you under the Apache License, Version 2.0 (the
007: * "License"); you may not use this file except in compliance
008: * with the License. You may obtain a copy of the License at
009: *
010: * http://www.apache.org/licenses/LICENSE-2.0
011: *
012: * Unless required by applicable law or agreed to in writing,
013: * software distributed under the License is distributed on an
014: * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
015: * KIND, either express or implied. See the License for the
016: * specific language governing permissions and limitations
017: * under the License.
018: */package org.apache.cxf.systest.ws.rm;
019:
020: import java.io.IOException;
021: import java.io.OutputStream;
022: import java.math.BigInteger;
023: import java.util.ListIterator;
024: import java.util.logging.Level;
025: import java.util.logging.Logger;
026:
027: import org.apache.cxf.interceptor.Fault;
028: import org.apache.cxf.interceptor.InterceptorChain;
029: import org.apache.cxf.interceptor.MessageSenderInterceptor;
030: import org.apache.cxf.io.AbstractWrappedOutputStream;
031: import org.apache.cxf.message.Message;
032: import org.apache.cxf.message.MessageUtils;
033: import org.apache.cxf.phase.AbstractPhaseInterceptor;
034: import org.apache.cxf.phase.Phase;
035: import org.apache.cxf.phase.PhaseInterceptor;
036: import org.apache.cxf.ws.addressing.AddressingProperties;
037: import org.apache.cxf.ws.rm.RMContextUtils;
038:
039: /**
040: *
041: */
042: public class MessageLossSimulator extends
043: AbstractPhaseInterceptor<Message> {
044:
045: private static final Logger LOG = Logger
046: .getLogger(MessageLossSimulator.class.getName());
047: private int appMessageCount;
048:
049: public MessageLossSimulator() {
050: super (Phase.PREPARE_SEND);
051: addBefore(MessageSenderInterceptor.class.getName());
052: }
053:
054: public void handleMessage(Message message) throws Fault {
055: AddressingProperties maps = RMContextUtils.retrieveMAPs(
056: message, false, true);
057: RMContextUtils.ensureExposedVersion(maps);
058: String action = null;
059: if (maps != null && null != maps.getAction()) {
060: action = maps.getAction().getValue();
061: }
062: if (RMContextUtils.isRMProtocolMessage(action)) {
063: return;
064: }
065: if (MessageUtils.isPartialResponse(message)) {
066: return;
067: }
068: appMessageCount++;
069: if (0 != (appMessageCount % 2)) {
070: return;
071: }
072:
073: InterceptorChain chain = message.getInterceptorChain();
074: ListIterator it = chain.getIterator();
075: while (it.hasNext()) {
076: PhaseInterceptor pi = (PhaseInterceptor) it.next();
077: if (MessageSenderInterceptor.class.getName().equals(
078: pi.getId())) {
079: chain.remove(pi);
080: LOG
081: .fine("Removed MessageSenderInterceptor from interceptor chain.");
082: break;
083: }
084: }
085:
086: message.setContent(OutputStream.class, new WrappedOutputStream(
087: message));
088:
089: message.getInterceptorChain().add(
090: new AbstractPhaseInterceptor<Message>(
091: Phase.PREPARE_SEND_ENDING) {
092:
093: public void handleMessage(Message message)
094: throws Fault {
095: try {
096: message.getContent(OutputStream.class)
097: .close();
098: } catch (IOException e) {
099: throw new Fault(e);
100: }
101: }
102:
103: });
104: }
105:
106: private class WrappedOutputStream extends
107: AbstractWrappedOutputStream {
108:
109: private Message outMessage;
110:
111: public WrappedOutputStream(Message m) {
112: this .outMessage = m;
113: }
114:
115: @Override
116: protected void onFirstWrite() throws IOException {
117: if (LOG.isLoggable(Level.FINE)) {
118: BigInteger nr = RMContextUtils.retrieveRMProperties(
119: outMessage, true).getSequence()
120: .getMessageNumber();
121: LOG.fine("Losing message " + nr);
122: }
123: wrappedStream = new DummyOutputStream();
124: }
125: }
126:
127: private class DummyOutputStream extends OutputStream {
128:
129: @Override
130: public void write(int b) throws IOException {
131: // TODO Auto-generated method stub
132:
133: }
134:
135: }
136:
137: }
|