View Javadoc

1   package org.apache.maven.plugin.surefire.booterclient.output;
2   
3   /*
4    * Licensed to the Apache Software Foundation (ASF) under one
5    * or more contributor license agreements.  See the NOTICE file
6    * distributed with this work for additional information
7    * regarding copyright ownership.  The ASF licenses this file
8    * to you under the Apache License, Version 2.0 (the
9    * "License"); you may not use this file except in compliance
10   * with the License.  You may obtain a copy of the License at
11   *
12   *     http://www.apache.org/licenses/LICENSE-2.0
13   *
14   * Unless required by applicable law or agreed to in writing,
15   * software distributed under the License is distributed on an
16   * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
17   * KIND, either express or implied.  See the License for the
18   * specific language governing permissions and limitations
19   * under the License.
20   */
21  
22  import org.apache.maven.surefire.util.internal.BlockingQueue;
23  import org.apache.maven.surefire.util.internal.BlockingQueueFactory;
24  import org.codehaus.plexus.util.cli.StreamConsumer;
25  
26  /**
27   * Knows how to reconstruct *all* the state transmitted over stdout by the forked process.
28   *
29   * @author Kristian Rosenvold
30   */
31  public class ThreadedStreamConsumer
32      implements StreamConsumer
33  {
34  
35      private final BlockingQueue items = BlockingQueueFactory.createBlockingQueue();
36  
37      private static final String poison = "Pioson";
38  
39      private final Thread thread;
40  
41      private final Pumper pumper;
42  
43      static class Pumper
44          implements Runnable
45      {
46          private final BlockingQueue queue;
47  
48          private final StreamConsumer target;
49  
50          private volatile InterruptedException interruptedException;
51  
52  
53          Pumper( BlockingQueue queue, StreamConsumer target )
54          {
55              this.queue = queue;
56              this.target = target;
57          }
58  
59          public void run()
60          {
61              try
62              {
63                  String item = (String) queue.take();
64                  //noinspection StringEquality
65                  while ( item != poison )
66                  {
67                      target.consumeLine( item );
68                      item = (String) queue.take();
69                  }
70              }
71              catch ( InterruptedException e )
72              {
73                  this.interruptedException = e;
74              }
75          }
76  
77          public InterruptedException getInterruptedException()
78          {
79              return interruptedException;
80          }
81      }
82  
83      public ThreadedStreamConsumer( StreamConsumer target )
84      {
85          pumper = new Pumper( items, target );
86          thread = new Thread( pumper, "ThreadedStreamConsumer" );
87          thread.start();
88      }
89  
90      public void consumeLine( String s )
91      {
92          items.add( s );
93      }
94  
95  
96      public void close()
97      {
98          try
99          {
100             items.add( poison );
101             thread.join();
102             //noinspection ThrowableResultOfMethodCallIgnored
103             if ( pumper.getInterruptedException() != null )
104             {
105                 throw pumper.getInterruptedException();
106             }
107         }
108         catch ( InterruptedException e )
109         {
110             throw new RuntimeException( e );
111         }
112     }
113 }