001package org.apache.maven.lifecycle.internal.builder.multithreaded;
002
003/*
004 * Licensed to the Apache Software Foundation (ASF) under one
005 * or more contributor license agreements.  See the NOTICE file
006 * distributed with this work for additional information
007 * regarding copyright ownership.  The ASF licenses this file
008 * to you under the Apache License, Version 2.0 (the
009 * "License"); you may not use this file except in compliance
010 * with the License.  You may obtain a copy of the License at
011 *
012 *  http://www.apache.org/licenses/LICENSE-2.0
013 *
014 * Unless required by applicable law or agreed to in writing,
015 * software distributed under the License is distributed on an
016 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
017 * KIND, either express or implied.  See the License for the
018 * specific language governing permissions and limitations
019 * under the License.
020 */
021
022import java.io.ByteArrayOutputStream;
023import java.io.IOException;
024import java.io.PrintStream;
025import java.util.Collections;
026import java.util.HashMap;
027import java.util.HashSet;
028import java.util.Iterator;
029import java.util.Map;
030import java.util.Set;
031
032import org.apache.maven.lifecycle.internal.ProjectBuildList;
033import org.apache.maven.lifecycle.internal.ProjectSegment;
034
035/**
036 * @since 3.0
037 * @author Kristian Rosenvold
038 *         <p/>
039 *         NOTE: This class is not part of any public api and can be changed or deleted without prior notice.
040 *         This class in particular may spontaneusly self-combust and be replaced by a plexus-compliant thread aware
041 *         logger implementation at any time.
042 */
043@SuppressWarnings( { "SynchronizationOnLocalVariableOrMethodParameter" } )
044public class ThreadOutputMuxer
045{
046    private final Iterator<ProjectSegment> projects;
047
048    private final ThreadLocal<ProjectSegment> projectBuildThreadLocal = new ThreadLocal<>();
049
050    private final Map<ProjectSegment, ByteArrayOutputStream> streams =
051        new HashMap<>();
052
053    private final Map<ProjectSegment, PrintStream> printStreams = new HashMap<>();
054
055    private final ByteArrayOutputStream defaultOutputStreamForUnknownData = new ByteArrayOutputStream();
056
057    private final PrintStream defaultPringStream = new PrintStream( defaultOutputStreamForUnknownData );
058
059    private final Set<ProjectSegment> completedBuilds = Collections.synchronizedSet( new HashSet<ProjectSegment>() );
060
061    private volatile ProjectSegment currentBuild;
062
063    private final PrintStream originalSystemOUtStream;
064
065    private final ConsolePrinter printer;
066
067    /**
068     * A simple but safe solution for printing to the console.
069     */
070
071    class ConsolePrinter
072        implements Runnable
073    {
074        public volatile boolean running;
075
076        private final ProjectBuildList projectBuildList;
077
078        ConsolePrinter( ProjectBuildList projectBuildList )
079        {
080            this.projectBuildList = projectBuildList;
081        }
082
083        public void run()
084        {
085            running = true;
086            for ( ProjectSegment projectBuild : projectBuildList )
087            {
088                final PrintStream projectStream = printStreams.get( projectBuild );
089                ByteArrayOutputStream projectOs = streams.get( projectBuild );
090
091                do
092                {
093                    synchronized ( projectStream )
094                    {
095                        try
096                        {
097                            projectStream.wait( 100 );
098                        }
099                        catch ( InterruptedException e )
100                        {
101                            throw new RuntimeException( e );
102                        }
103                        try
104                        {
105                            projectOs.writeTo( originalSystemOUtStream );
106                        }
107                        catch ( IOException e )
108                        {
109                            throw new RuntimeException( e );
110                        }
111
112                        projectOs.reset();
113                    }
114                }
115                while ( !completedBuilds.contains( projectBuild ) );
116            }
117            running = false;
118        }
119
120        /*
121        Wait until we are sure the print-stream thread is running.
122         */
123
124        public void waitUntilRunning( boolean expect )
125        {
126            while ( !running == expect )
127            {
128                try
129                {
130                    Thread.sleep( 10 );
131                }
132                catch ( InterruptedException e )
133                {
134                    throw new RuntimeException( e );
135                }
136            }
137        }
138    }
139
140    public ThreadOutputMuxer( ProjectBuildList segmentChunks, PrintStream originalSystemOut )
141    {
142        projects = segmentChunks.iterator();
143        for ( ProjectSegment segmentChunk : segmentChunks )
144        {
145            final ByteArrayOutputStream value = new ByteArrayOutputStream();
146            streams.put( segmentChunk, value );
147            printStreams.put( segmentChunk, new PrintStream( value ) );
148        }
149        setNext();
150        this.originalSystemOUtStream = originalSystemOut;
151        System.setOut( new ThreadBoundPrintStream( this.originalSystemOUtStream ) );
152        printer = new ConsolePrinter( segmentChunks );
153        new Thread( printer ).start();
154        printer.waitUntilRunning( true );
155    }
156
157    public void close()
158    {
159        printer.waitUntilRunning( false );
160        System.setOut( this.originalSystemOUtStream );
161    }
162
163    private void setNext()
164    {
165        currentBuild = projects.hasNext() ? projects.next() : null;
166    }
167
168    private boolean ownsRealOutputStream( ProjectSegment projectBuild )
169    {
170        return projectBuild.equals( currentBuild );
171    }
172
173    private PrintStream getThreadBoundPrintStream()
174    {
175        ProjectSegment threadProject = projectBuildThreadLocal.get();
176        if ( threadProject == null )
177        {
178            return defaultPringStream;
179        }
180        if ( ownsRealOutputStream( threadProject ) )
181        {
182            return originalSystemOUtStream;
183        }
184        return printStreams.get( threadProject );
185    }
186
187    public void associateThreadWithProjectSegment( ProjectSegment projectBuild )
188    {
189        projectBuildThreadLocal.set( projectBuild );
190    }
191
192    public void setThisModuleComplete( ProjectSegment projectBuild )
193    {
194        completedBuilds.add( projectBuild );
195        PrintStream stream = printStreams.get( projectBuild );
196        synchronized ( stream )
197        {
198            stream.notifyAll();
199        }
200        disconnectThreadFromProject();
201    }
202
203    private void disconnectThreadFromProject()
204    {
205        projectBuildThreadLocal.remove();
206    }
207
208    private class ThreadBoundPrintStream
209        extends PrintStream
210    {
211
212        public ThreadBoundPrintStream( PrintStream systemOutStream )
213        {
214            super( systemOutStream );
215        }
216
217        private PrintStream getOutputStreamForCurrentThread()
218        {
219            return getThreadBoundPrintStream();
220        }
221
222        @Override
223        public void println()
224        {
225            final PrintStream currentStream = getOutputStreamForCurrentThread();
226            synchronized ( currentStream )
227            {
228                currentStream.println();
229                currentStream.notifyAll();
230            }
231        }
232
233        @Override
234        public void print( char c )
235        {
236            final PrintStream currentStream = getOutputStreamForCurrentThread();
237            synchronized ( currentStream )
238            {
239                currentStream.print( c );
240                currentStream.notifyAll();
241            }
242        }
243
244        @Override
245        public void println( char x )
246        {
247            final PrintStream currentStream = getOutputStreamForCurrentThread();
248            synchronized ( currentStream )
249            {
250                currentStream.println( x );
251                currentStream.notifyAll();
252            }
253        }
254
255        @Override
256        public void print( double d )
257        {
258            final PrintStream currentStream = getOutputStreamForCurrentThread();
259            synchronized ( currentStream )
260            {
261                currentStream.print( d );
262                currentStream.notifyAll();
263            }
264        }
265
266        @Override
267        public void println( double x )
268        {
269            final PrintStream currentStream = getOutputStreamForCurrentThread();
270            synchronized ( currentStream )
271            {
272                currentStream.println( x );
273                currentStream.notifyAll();
274            }
275        }
276
277        @Override
278        public void print( float f )
279        {
280            final PrintStream currentStream = getOutputStreamForCurrentThread();
281            synchronized ( currentStream )
282            {
283                currentStream.print( f );
284                currentStream.notifyAll();
285            }
286        }
287
288        @Override
289        public void println( float x )
290        {
291            final PrintStream currentStream = getOutputStreamForCurrentThread();
292            synchronized ( currentStream )
293            {
294                currentStream.println( x );
295                currentStream.notifyAll();
296            }
297        }
298
299        @Override
300        public void print( int i )
301        {
302            final PrintStream currentStream = getOutputStreamForCurrentThread();
303            synchronized ( currentStream )
304            {
305                currentStream.print( i );
306                currentStream.notifyAll();
307            }
308        }
309
310        @Override
311        public void println( int x )
312        {
313            final PrintStream currentStream = getOutputStreamForCurrentThread();
314            synchronized ( currentStream )
315            {
316                currentStream.println( x );
317                currentStream.notifyAll();
318            }
319        }
320
321        @Override
322        public void print( long l )
323        {
324            final PrintStream currentStream = getOutputStreamForCurrentThread();
325            synchronized ( currentStream )
326            {
327                currentStream.print( l );
328                currentStream.notifyAll();
329            }
330        }
331
332        @Override
333        public void println( long x )
334        {
335            final PrintStream currentStream = getOutputStreamForCurrentThread();
336            synchronized ( currentStream )
337            {
338                currentStream.print( x );
339                currentStream.notifyAll();
340            }
341        }
342
343        @Override
344        public void print( boolean b )
345        {
346            final PrintStream currentStream = getOutputStreamForCurrentThread();
347            synchronized ( currentStream )
348            {
349                currentStream.print( b );
350                currentStream.notifyAll();
351            }
352        }
353
354        @Override
355        public void println( boolean x )
356        {
357            final PrintStream currentStream = getOutputStreamForCurrentThread();
358            synchronized ( currentStream )
359            {
360                currentStream.print( x );
361                currentStream.notifyAll();
362            }
363        }
364
365        @Override
366        public void print( char s[] )
367        {
368            final PrintStream currentStream = getOutputStreamForCurrentThread();
369            synchronized ( currentStream )
370            {
371                currentStream.print( s );
372                currentStream.notifyAll();
373            }
374        }
375
376        @Override
377        public void println( char x[] )
378        {
379            final PrintStream currentStream = getOutputStreamForCurrentThread();
380            synchronized ( currentStream )
381            {
382                currentStream.print( x );
383                currentStream.notifyAll();
384            }
385        }
386
387        @Override
388        public void print( Object obj )
389        {
390            final PrintStream currentStream = getOutputStreamForCurrentThread();
391            synchronized ( currentStream )
392            {
393                currentStream.print( obj );
394                currentStream.notifyAll();
395            }
396        }
397
398        @Override
399        public void println( Object x )
400        {
401            final PrintStream currentStream = getOutputStreamForCurrentThread();
402            synchronized ( currentStream )
403            {
404                currentStream.println( x );
405                currentStream.notifyAll();
406            }
407        }
408
409        @Override
410        public void print( String s )
411        {
412            final PrintStream currentStream = getOutputStreamForCurrentThread();
413            synchronized ( currentStream )
414            {
415                currentStream.print( s );
416                currentStream.notifyAll();
417            }
418        }
419
420        @Override
421        public void println( String x )
422        {
423            final PrintStream currentStream = getOutputStreamForCurrentThread();
424            synchronized ( currentStream )
425            {
426                currentStream.println( x );
427                currentStream.notifyAll();
428            }
429        }
430
431        @Override
432        public void write( byte b[], int off, int len )
433        {
434            final PrintStream currentStream = getOutputStreamForCurrentThread();
435            synchronized ( currentStream )
436            {
437                currentStream.write( b, off, len );
438                currentStream.notifyAll();
439            }
440        }
441
442        @Override
443        public void close()
444        {
445            getOutputStreamForCurrentThread().close();
446        }
447
448        @Override
449        public void flush()
450        {
451            getOutputStreamForCurrentThread().flush();
452        }
453
454        @Override
455        public void write( int b )
456        {
457            final PrintStream currentStream = getOutputStreamForCurrentThread();
458            synchronized ( currentStream )
459            {
460                currentStream.write( b );
461                currentStream.notifyAll();
462            }
463        }
464
465        @Override
466        public void write( byte b[] )
467            throws IOException
468        {
469            final PrintStream currentStream = getOutputStreamForCurrentThread();
470            synchronized ( currentStream )
471            {
472                currentStream.write( b );
473                currentStream.notifyAll();
474            }
475        }
476    }
477}