diff options
Diffstat (limited to 'plugins/org.eclipse.net4j')
91 files changed, 6515 insertions, 425 deletions
diff --git a/plugins/org.eclipse.net4j/.classpath b/plugins/org.eclipse.net4j/.classpath new file mode 100644 index 0000000000..8ea6ca0559 --- /dev/null +++ b/plugins/org.eclipse.net4j/.classpath @@ -0,0 +1,7 @@ +<?xml version="1.0" encoding="UTF-8"?> +<classpath> + <classpathentry excluding="org/eclipse/internal/net4j/bundle/Activator.java|org/eclipse/internal/net4j/bundle/Log.java|org/eclipse/internal/net4j/component/InactiveException.java|org/eclipse/net4j/transport/ChannelMultiplexer.java" kind="src" path="src"/> + <classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER/org.eclipse.jdt.internal.debug.ui.launcher.StandardVMType/J2SE-1.5"/> + <classpathentry kind="con" path="org.eclipse.pde.core.requiredPlugins"/> + <classpathentry kind="output" path="bin"/> +</classpath> diff --git a/plugins/org.eclipse.net4j/.cvsignore b/plugins/org.eclipse.net4j/.cvsignore new file mode 100644 index 0000000000..ba077a4031 --- /dev/null +++ b/plugins/org.eclipse.net4j/.cvsignore @@ -0,0 +1 @@ +bin diff --git a/plugins/org.eclipse.net4j/.project b/plugins/org.eclipse.net4j/.project index 6ee99bc561..fd0305df1a 100644 --- a/plugins/org.eclipse.net4j/.project +++ b/plugins/org.eclipse.net4j/.project @@ -6,6 +6,11 @@ </projects> <buildSpec> <buildCommand> + <name>org.eclipse.jdt.core.javabuilder</name> + <arguments> + </arguments> + </buildCommand> + <buildCommand> <name>org.eclipse.pde.ManifestBuilder</name> <arguments> </arguments> @@ -18,5 +23,6 @@ </buildSpec> <natures> <nature>org.eclipse.pde.PluginNature</nature> + <nature>org.eclipse.jdt.core.javanature</nature> </natures> </projectDescription> diff --git a/plugins/org.eclipse.net4j/.settings/org.eclipse.jdt.core.prefs b/plugins/org.eclipse.net4j/.settings/org.eclipse.jdt.core.prefs new file mode 100644 index 0000000000..cedc7e0052 --- /dev/null +++ b/plugins/org.eclipse.net4j/.settings/org.eclipse.jdt.core.prefs @@ -0,0 +1,7 @@ +#Fri Oct 20 10:25:00 CEST 2006 +eclipse.preferences.version=1 +org.eclipse.jdt.core.compiler.codegen.targetPlatform=1.5 +org.eclipse.jdt.core.compiler.compliance=1.5 +org.eclipse.jdt.core.compiler.problem.assertIdentifier=error +org.eclipse.jdt.core.compiler.problem.enumIdentifier=error +org.eclipse.jdt.core.compiler.source=1.5 diff --git a/plugins/org.eclipse.net4j/.settings/org.eclipse.pde.core.prefs b/plugins/org.eclipse.net4j/.settings/org.eclipse.pde.core.prefs new file mode 100644 index 0000000000..794bf7cd88 --- /dev/null +++ b/plugins/org.eclipse.net4j/.settings/org.eclipse.pde.core.prefs @@ -0,0 +1,3 @@ +#Sun Oct 15 23:11:17 CEST 2006 +eclipse.preferences.version=1 +pluginProject.extensions=false diff --git a/plugins/org.eclipse.net4j/META-INF/MANIFEST.MF b/plugins/org.eclipse.net4j/META-INF/MANIFEST.MF index e63d9d1f35..7294a1e280 100644 --- a/plugins/org.eclipse.net4j/META-INF/MANIFEST.MF +++ b/plugins/org.eclipse.net4j/META-INF/MANIFEST.MF @@ -1,8 +1,21 @@ Manifest-Version: 1.0 Bundle-ManifestVersion: 2 -Bundle-Name: %pluginName +Bundle-Name: Net4j Plug-in Bundle-SymbolicName: org.eclipse.net4j -Bundle-Version: 0.7.0.qualifier -Bundle-Vendor: %providerName +Bundle-Version: 0.8.0.qualifier Bundle-Localization: plugin -Eclipse-LazyStart: true +Bundle-RequiredExecutionEnvironment: J2SE-1.5 +Import-Package: org.osgi.framework;version="1.3.0" +Export-Package: org.eclipse.internal.net4j.transport;version="0.8.0", + org.eclipse.internal.net4j.transport.embedded;version="0.8.0", + org.eclipse.internal.net4j.transport.tcp;version="0.8.0", + org.eclipse.net4j.message;version="0.8.0", + org.eclipse.net4j.remote;version="0.8.0", + org.eclipse.net4j.signal;version="0.8.0", + org.eclipse.net4j.transport;version="0.8.0", + org.eclipse.net4j.transport.tcp;version="0.8.0", + org.eclipse.net4j.util.concurrent;version="0.8.0", + org.eclipse.net4j.util.lifecycle;version="0.8.0", + org.eclipse.net4j.util.map;version="0.8.0", + org.eclipse.net4j.util.registry;version="0.8.0", + org.eclipse.net4j.util.stream;version="0.8.0" diff --git a/plugins/org.eclipse.net4j/TODOS.txt b/plugins/org.eclipse.net4j/TODOS.txt new file mode 100644 index 0000000000..e69de29bb2 --- /dev/null +++ b/plugins/org.eclipse.net4j/TODOS.txt diff --git a/plugins/org.eclipse.net4j/about.html b/plugins/org.eclipse.net4j/about.html deleted file mode 100644 index 4c99086f82..0000000000 --- a/plugins/org.eclipse.net4j/about.html +++ /dev/null @@ -1,22 +0,0 @@ -<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.0//EN"> -<html> -<head> -<meta http-equiv=Content-Type content="text/html; charset=ISO-8859-1"> -<title>About</title> -</head> -<body lang="EN-US"> -<h2>About This Content</h2> - -<p>February 24, 2005</p> -<h3>License</h3> - -<p>The Eclipse Foundation makes available all content in this plug-in ("Content"). Unless otherwise indicated below, the Content is provided to you under the terms and conditions of the -Eclipse Public License Version 1.0 ("EPL"). A copy of the EPL is available at <a href="http://www.eclipse.org/legal/epl-v10.html">http://www.eclipse.org/legal/epl-v10.html</a>. -For purposes of the EPL, "Program" will mean the Content.</p> - -<p>If you did not receive this Content directly from the Eclipse Foundation, the Content is being redistributed by another party ("Redistributor") and different terms and conditions may -apply to your use of any object code in the Content. Check the Redistributor's license that was provided with the Content. If no such license exists, contact the Redistributor. Unless otherwise -indicated below, the terms and conditions of the EPL still apply to any source code in the Content.</p> - -</body> -</html>
\ No newline at end of file diff --git a/plugins/org.eclipse.net4j/about.ini b/plugins/org.eclipse.net4j/about.ini deleted file mode 100644 index c96a946248..0000000000 --- a/plugins/org.eclipse.net4j/about.ini +++ /dev/null @@ -1,17 +0,0 @@ -# about.ini -# contains information about a feature -# java.io.Properties file (ISO 8859-1 with "\" escapes) -# "%key" are externalized strings defined in about.properties -# This file does not need to be translated. - -# Property "aboutText" contains blurb for "About" dialog (translated) -aboutText=%featureText - -# Property "featureImage" contains path to feature image (32x32) -featureImage=eclipse32.png - -# Property "appName" contains name of the application (translated) -appName=%featureName - -# Property "welcomePage" contains path to welcome page (special XML-based format) -welcomePage=$nl$/welcome.xml diff --git a/plugins/org.eclipse.net4j/about.mappings b/plugins/org.eclipse.net4j/about.mappings deleted file mode 100644 index a28390a75e..0000000000 --- a/plugins/org.eclipse.net4j/about.mappings +++ /dev/null @@ -1,6 +0,0 @@ -# about.mappings -# contains fill-ins for about.properties -# java.io.Properties file (ISO 8859-1 with "\" escapes) -# This file does not need to be translated. - -0=@build@ diff --git a/plugins/org.eclipse.net4j/about.properties b/plugins/org.eclipse.net4j/about.properties deleted file mode 100644 index ebc7f9e982..0000000000 --- a/plugins/org.eclipse.net4j/about.properties +++ /dev/null @@ -1,11 +0,0 @@ -# NLS_MESSAGEFORMAT_VAR - -featureName=Net4j - -featureText=Net4j \n\ -Version: {featureVersion}\n\ -Build id: {0}\n\ -\n\ -Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. All rights reserved.\n\ -\n\ -Visit http://www.eclipse.org/emft/projects/net4j diff --git a/plugins/org.eclipse.net4j/build.properties b/plugins/org.eclipse.net4j/build.properties index dd160d114d..34d2e4d2da 100644 --- a/plugins/org.eclipse.net4j/build.properties +++ b/plugins/org.eclipse.net4j/build.properties @@ -1,10 +1,4 @@ -# NLS_MESSAGEFORMAT_VAR -bin.includes = about.html,\ - about.ini,\ - about.mappings,\ - about.properties,\ - plugin.xml,\ - plugin.properties,\ - META-INF/,\ - eclipse32.png,\ - license.html +source.. = src/ +output.. = bin/ +bin.includes = META-INF/,\ + . diff --git a/plugins/org.eclipse.net4j/eclipse32.png b/plugins/org.eclipse.net4j/eclipse32.png Binary files differdeleted file mode 100644 index 568fac1d05..0000000000 --- a/plugins/org.eclipse.net4j/eclipse32.png +++ /dev/null diff --git a/plugins/org.eclipse.net4j/javadoc.xml b/plugins/org.eclipse.net4j/javadoc.xml new file mode 100644 index 0000000000..52a4c5049d --- /dev/null +++ b/plugins/org.eclipse.net4j/javadoc.xml @@ -0,0 +1,31 @@ +<?xml version="1.0" encoding="UTF-8"?> +<project default="javadoc"> + <target name="javadoc"> + <javadoc + access="protected" + author="true" + destdir="doc" + nodeprecated="false" + nodeprecatedlist="false" + noindex="false" + nonavbar="false" + notree="false" + source="1.5" + sourcepath="src" + splitindex="true" + use="true" + version="true"> + <package name="org.eclipse.net4j"/> + <package name="org.eclipse.net4j.message"/> + <package name="org.eclipse.net4j.remote"/> + <package name="org.eclipse.net4j.signal"/> + <package name="org.eclipse.net4j.transport"/> + <package name="org.eclipse.net4j.transport.tcp"/> + <package name="org.eclipse.net4j.util"/> + <package name="org.eclipse.net4j.util.concurrent"/> + <package name="org.eclipse.net4j.util.lifecycle"/> + <package name="org.eclipse.net4j.util.registry"/> + <link href="http://java.sun.com/j2se/1.5.0/docs/api/" /> + </javadoc> + </target> +</project> diff --git a/plugins/org.eclipse.net4j/license.html b/plugins/org.eclipse.net4j/license.html deleted file mode 100644 index d7b88e9416..0000000000 --- a/plugins/org.eclipse.net4j/license.html +++ /dev/null @@ -1,319 +0,0 @@ -<html xmlns:o="urn:schemas-microsoft-com:office:office" xmlns:w="urn:schemas-microsoft-com:office:word" xmlns="http://www.w3.org/TR/REC-html40"><head> -<meta http-equiv="Content-Type" content="text/html; charset=windows-1252"> -<meta name="ProgId" content="Word.Document"> -<meta name="Generator" content="Microsoft Word 9"> -<meta name="Originator" content="Microsoft Word 9"> -<link rel="File-List" href="http://www.eclipse.org/org/documents/Eclipse%20EPL%202003_11_10%20Final_files/filelist.xml"><title>Eclipse Public License - Version 1.0</title><!--[if gte mso 9]><xml> - <o:DocumentProperties> - <o:Revision>2</o:Revision> - <o:TotalTime>3</o:TotalTime> - <o:Created>2004-03-05T23:03:00Z</o:Created> - <o:LastSaved>2004-03-05T23:03:00Z</o:LastSaved> - <o:Pages>4</o:Pages> - <o:Words>1626</o:Words> - <o:Characters>9270</o:Characters> - <o:Lines>77</o:Lines> - <o:Paragraphs>18</o:Paragraphs> - <o:CharactersWithSpaces>11384</o:CharactersWithSpaces> - <o:Version>9.4402</o:Version> - </o:DocumentProperties> -</xml><![endif]--><!--[if gte mso 9]><xml> - <w:WordDocument> - <w:TrackRevisions/> - </w:WordDocument> -</xml><![endif]--> - - -<style> -<!-- - /* Font Definitions */ -@font-face - {font-family:Tahoma; - panose-1:2 11 6 4 3 5 4 4 2 4; - mso-font-charset:0; - mso-generic-font-family:swiss; - mso-font-pitch:variable; - mso-font-signature:553679495 -2147483648 8 0 66047 0;} - /* Style Definitions */ -p.MsoNormal, li.MsoNormal, div.MsoNormal - {mso-style-parent:""; - margin:0in; - margin-bottom:.0001pt; - mso-pagination:widow-orphan; - font-size:12.0pt; - font-family:"Times New Roman"; - mso-fareast-font-family:"Times New Roman";} -p - {margin-right:0in; - mso-margin-top-alt:auto; - mso-margin-bottom-alt:auto; - margin-left:0in; - mso-pagination:widow-orphan; - font-size:12.0pt; - font-family:"Times New Roman"; - mso-fareast-font-family:"Times New Roman";} -p.BalloonText, li.BalloonText, div.BalloonText - {mso-style-name:"Balloon Text"; - margin:0in; - margin-bottom:.0001pt; - mso-pagination:widow-orphan; - font-size:8.0pt; - font-family:Tahoma; - mso-fareast-font-family:"Times New Roman";} -@page Section1 - {size:8.5in 11.0in; - margin:1.0in 1.25in 1.0in 1.25in; - mso-header-margin:.5in; - mso-footer-margin:.5in; - mso-paper-source:0;} -div.Section1 - {page:Section1;} ---> -</style></head> - -<body style="" lang="EN-US"> - -<div class="Section1"> - -<p style="text-align: center;" align="center"><b>Eclipse Public License - v 1.0</b> -</p> - -<p><span style="font-size: 10pt;">THE ACCOMPANYING PROGRAM IS PROVIDED UNDER -THE TERMS OF THIS ECLIPSE PUBLIC LICENSE ("AGREEMENT"). ANY USE, -REPRODUCTION OR DISTRIBUTION OF THE PROGRAM CONSTITUTES RECIPIENT'S ACCEPTANCE -OF THIS AGREEMENT.</span> </p> - -<p><b><span style="font-size: 10pt;">1. DEFINITIONS</span></b> </p> - -<p><span style="font-size: 10pt;">"Contribution" means:</span> </p> - -<p class="MsoNormal" style="margin-left: 0.5in;"><span style="font-size: 10pt;">a) -in the case of the initial Contributor, the initial code and documentation -distributed under this Agreement, and<br clear="left"> -b) in the case of each subsequent Contributor:</span></p> - -<p class="MsoNormal" style="margin-left: 0.5in;"><span style="font-size: 10pt;">i) -changes to the Program, and</span></p> - -<p class="MsoNormal" style="margin-left: 0.5in;"><span style="font-size: 10pt;">ii) -additions to the Program;</span></p> - -<p class="MsoNormal" style="margin-left: 0.5in;"><span style="font-size: 10pt;">where -such changes and/or additions to the Program originate from and are distributed -by that particular Contributor. A Contribution 'originates' from a Contributor -if it was added to the Program by such Contributor itself or anyone acting on -such Contributor's behalf. Contributions do not include additions to the -Program which: (i) are separate modules of software distributed in conjunction -with the Program under their own license agreement, and (ii) are not derivative -works of the Program. </span></p> - -<p><span style="font-size: 10pt;">"Contributor" means any person or -entity that distributes the Program.</span> </p> - -<p><span style="font-size: 10pt;">"Licensed Patents " mean patent -claims licensable by a Contributor which are necessarily infringed by the use -or sale of its Contribution alone or when combined with the Program. </span></p> - -<p><span style="font-size: 10pt;">"Program" means the Contributions -distributed in accordance with this Agreement.</span> </p> - -<p><span style="font-size: 10pt;">"Recipient" means anyone who -receives the Program under this Agreement, including all Contributors.</span> </p> - -<p><b><span style="font-size: 10pt;">2. GRANT OF RIGHTS</span></b> </p> - -<p class="MsoNormal" style="margin-left: 0.5in;"><span style="font-size: 10pt;">a) -Subject to the terms of this Agreement, each Contributor hereby grants Recipient -a non-exclusive, worldwide, royalty-free copyright license to<span style="color: red;"> </span>reproduce, prepare derivative works of, publicly -display, publicly perform, distribute and sublicense the Contribution of such -Contributor, if any, and such derivative works, in source code and object code -form.</span></p> - -<p class="MsoNormal" style="margin-left: 0.5in;"><span style="font-size: 10pt;">b) -Subject to the terms of this Agreement, each Contributor hereby grants -Recipient a non-exclusive, worldwide,<span style="color: green;"> </span>royalty-free -patent license under Licensed Patents to make, use, sell, offer to sell, import -and otherwise transfer the Contribution of such Contributor, if any, in source -code and object code form. This patent license shall apply to the combination -of the Contribution and the Program if, at the time the Contribution is added -by the Contributor, such addition of the Contribution causes such combination -to be covered by the Licensed Patents. The patent license shall not apply to -any other combinations which include the Contribution. No hardware per se is -licensed hereunder. </span></p> - -<p class="MsoNormal" style="margin-left: 0.5in;"><span style="font-size: 10pt;">c) -Recipient understands that although each Contributor grants the licenses to its -Contributions set forth herein, no assurances are provided by any Contributor -that the Program does not infringe the patent or other intellectual property -rights of any other entity. Each Contributor disclaims any liability to Recipient -for claims brought by any other entity based on infringement of intellectual -property rights or otherwise. As a condition to exercising the rights and -licenses granted hereunder, each Recipient hereby assumes sole responsibility -to secure any other intellectual property rights needed, if any. For example, -if a third party patent license is required to allow Recipient to distribute -the Program, it is Recipient's responsibility to acquire that license before -distributing the Program.</span></p> - -<p class="MsoNormal" style="margin-left: 0.5in;"><span style="font-size: 10pt;">d) -Each Contributor represents that to its knowledge it has sufficient copyright -rights in its Contribution, if any, to grant the copyright license set forth in -this Agreement. </span></p> - -<p><b><span style="font-size: 10pt;">3. REQUIREMENTS</span></b> </p> - -<p><span style="font-size: 10pt;">A Contributor may choose to distribute the -Program in object code form under its own license agreement, provided that:</span> -</p> - -<p class="MsoNormal" style="margin-left: 0.5in;"><span style="font-size: 10pt;">a) -it complies with the terms and conditions of this Agreement; and</span></p> - -<p class="MsoNormal" style="margin-left: 0.5in;"><span style="font-size: 10pt;">b) -its license agreement:</span></p> - -<p class="MsoNormal" style="margin-left: 0.5in;"><span style="font-size: 10pt;">i) -effectively disclaims on behalf of all Contributors all warranties and -conditions, express and implied, including warranties or conditions of title -and non-infringement, and implied warranties or conditions of merchantability -and fitness for a particular purpose; </span></p> - -<p class="MsoNormal" style="margin-left: 0.5in;"><span style="font-size: 10pt;">ii) -effectively excludes on behalf of all Contributors all liability for damages, -including direct, indirect, special, incidental and consequential damages, such -as lost profits; </span></p> - -<p class="MsoNormal" style="margin-left: 0.5in;"><span style="font-size: 10pt;">iii) -states that any provisions which differ from this Agreement are offered by that -Contributor alone and not by any other party; and</span></p> - -<p class="MsoNormal" style="margin-left: 0.5in;"><span style="font-size: 10pt;">iv) -states that source code for the Program is available from such Contributor, and -informs licensees how to obtain it in a reasonable manner on or through a -medium customarily used for software exchange.<span style="color: blue;"> </span></span></p> - -<p><span style="font-size: 10pt;">When the Program is made available in source -code form:</span> </p> - -<p class="MsoNormal" style="margin-left: 0.5in;"><span style="font-size: 10pt;">a) -it must be made available under this Agreement; and </span></p> - -<p class="MsoNormal" style="margin-left: 0.5in;"><span style="font-size: 10pt;">b) a -copy of this Agreement must be included with each copy of the Program. </span></p> - -<p><span style="font-size: 10pt;">Contributors may not remove or alter any -copyright notices contained within the Program. </span></p> - -<p><span style="font-size: 10pt;">Each Contributor must identify itself as the -originator of its Contribution, if any, in a manner that reasonably allows -subsequent Recipients to identify the originator of the Contribution. </span></p> - -<p><b><span style="font-size: 10pt;">4. COMMERCIAL DISTRIBUTION</span></b> </p> - -<p><span style="font-size: 10pt;">Commercial distributors of software may -accept certain responsibilities with respect to end users, business partners -and the like. While this license is intended to facilitate the commercial use -of the Program, the Contributor who includes the Program in a commercial -product offering should do so in a manner which does not create potential -liability for other Contributors. Therefore, if a Contributor includes the -Program in a commercial product offering, such Contributor ("Commercial -Contributor") hereby agrees to defend and indemnify every other -Contributor ("Indemnified Contributor") against any losses, damages and -costs (collectively "Losses") arising from claims, lawsuits and other -legal actions brought by a third party against the Indemnified Contributor to -the extent caused by the acts or omissions of such Commercial Contributor in -connection with its distribution of the Program in a commercial product -offering. The obligations in this section do not apply to any claims or Losses -relating to any actual or alleged intellectual property infringement. In order -to qualify, an Indemnified Contributor must: a) promptly notify the Commercial -Contributor in writing of such claim, and b) allow the Commercial Contributor -to control, and cooperate with the Commercial Contributor in, the defense and -any related settlement negotiations. The Indemnified Contributor may participate -in any such claim at its own expense.</span> </p> - -<p><span style="font-size: 10pt;">For example, a Contributor might include the -Program in a commercial product offering, Product X. That Contributor is then a -Commercial Contributor. If that Commercial Contributor then makes performance -claims, or offers warranties related to Product X, those performance claims and -warranties are such Commercial Contributor's responsibility alone. Under this -section, the Commercial Contributor would have to defend claims against the -other Contributors related to those performance claims and warranties, and if a -court requires any other Contributor to pay any damages as a result, the -Commercial Contributor must pay those damages.</span> </p> - -<p><b><span style="font-size: 10pt;">5. NO WARRANTY</span></b> </p> - -<p><span style="font-size: 10pt;">EXCEPT AS EXPRESSLY SET FORTH IN THIS -AGREEMENT, THE PROGRAM IS PROVIDED ON AN "AS IS" BASIS, WITHOUT -WARRANTIES OR CONDITIONS OF ANY KIND, EITHER EXPRESS OR IMPLIED INCLUDING, -WITHOUT LIMITATION, ANY WARRANTIES OR CONDITIONS OF TITLE, NON-INFRINGEMENT, -MERCHANTABILITY OR FITNESS FOR A PARTICULAR PURPOSE. Each Recipient is solely -responsible for determining the appropriateness of using and distributing the -Program and assumes all risks associated with its exercise of rights under this -Agreement , including but not limited to the risks and costs of program errors, -compliance with applicable laws, damage to or loss of data, programs or -equipment, and unavailability or interruption of operations. </span></p> - -<p><b><span style="font-size: 10pt;">6. DISCLAIMER OF LIABILITY</span></b> </p> - -<p><span style="font-size: 10pt;">EXCEPT AS EXPRESSLY SET FORTH IN THIS -AGREEMENT, NEITHER RECIPIENT NOR ANY CONTRIBUTORS SHALL HAVE ANY LIABILITY FOR -ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES -(INCLUDING WITHOUT LIMITATION LOST PROFITS), HOWEVER CAUSED AND ON ANY THEORY -OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING -NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OR DISTRIBUTION OF -THE PROGRAM OR THE EXERCISE OF ANY RIGHTS GRANTED HEREUNDER, EVEN IF ADVISED OF -THE POSSIBILITY OF SUCH DAMAGES.</span> </p> - -<p><b><span style="font-size: 10pt;">7. GENERAL</span></b> </p> - -<p><span style="font-size: 10pt;">If any provision of this Agreement is invalid -or unenforceable under applicable law, it shall not affect the validity or -enforceability of the remainder of the terms of this Agreement, and without -further action by the parties hereto, such provision shall be reformed to the -minimum extent necessary to make such provision valid and enforceable.</span> </p> - -<p><span style="font-size: 10pt;">If Recipient institutes patent litigation -against any entity (including a cross-claim or counterclaim in a lawsuit) -alleging that the Program itself (excluding combinations of the Program with -other software or hardware) infringes such Recipient's patent(s), then such -Recipient's rights granted under Section 2(b) shall terminate as of the date -such litigation is filed. </span></p> - -<p><span style="font-size: 10pt;">All Recipient's rights under this Agreement -shall terminate if it fails to comply with any of the material terms or -conditions of this Agreement and does not cure such failure in a reasonable -period of time after becoming aware of such noncompliance. If all Recipient's -rights under this Agreement terminate, Recipient agrees to cease use and -distribution of the Program as soon as reasonably practicable. However, -Recipient's obligations under this Agreement and any licenses granted by -Recipient relating to the Program shall continue and survive. </span></p> - -<p><span style="font-size: 10pt;">Everyone is permitted to copy and distribute -copies of this Agreement, but in order to avoid inconsistency the Agreement is -copyrighted and may only be modified in the following manner. The Agreement -Steward reserves the right to publish new versions (including revisions) of -this Agreement from time to time. No one other than the Agreement Steward has -the right to modify this Agreement. The Eclipse Foundation is the initial -Agreement Steward. The Eclipse Foundation may assign the responsibility to -serve as the Agreement Steward to a suitable separate entity. Each new version -of the Agreement will be given a distinguishing version number. The Program -(including Contributions) may always be distributed subject to the version of -the Agreement under which it was received. In addition, after a new version of -the Agreement is published, Contributor may elect to distribute the Program -(including its Contributions) under the new version. Except as expressly stated -in Sections 2(a) and 2(b) above, Recipient receives no rights or licenses to -the intellectual property of any Contributor under this Agreement, whether -expressly, by implication, estoppel or otherwise. All rights in the Program not -expressly granted under this Agreement are reserved.</span> </p> - -<p><span style="font-size: 10pt;">This Agreement is governed by the laws of the -State of New York and the intellectual property laws of the United States of -America. No party to this Agreement will bring a legal action under this -Agreement more than one year after the cause of action arose. Each party waives -its rights to a jury trial in any resulting litigation.</span> </p> - -<p class="MsoNormal"><!--[if !supportEmptyParas]--> <!--[endif]--><o:p></o:p></p> - -</div> - -</body></html>
\ No newline at end of file diff --git a/plugins/org.eclipse.net4j/plugin.properties b/plugins/org.eclipse.net4j/plugin.properties deleted file mode 100644 index ad02e2ded7..0000000000 --- a/plugins/org.eclipse.net4j/plugin.properties +++ /dev/null @@ -1,33 +0,0 @@ -# /** -# * <copyright> -# * -# * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. -# * All rights reserved. This program and the accompanying materials -# * are made available under the terms of the Eclipse Public License v1.0 -# * which accompanies this distribution, and is available at -# * http://www.eclipse.org/legal/epl-v10.html -# * -# * Contributors: -# * Eike Stepper - Initial API and implementation -# * -# * </copyright> -# * -# * $Id$ -# */ - -# NLS_MESSAGEFORMAT_VAR - -# ============================================================================== -# Do not change the properties between this line and the last line containing: -# %%% END OF TRANSLATED PROPERTIES %%% -# Instead, either redefine an existing property, or create a new property, -# append it to the end of the file, and change the code to use the new name. -# ============================================================================== - -pluginName=Net4j -providerName=Eclipse.org - -# ============================================================================== -# %%% END OF TRANSLATED PROPERTIES %%% -# The above properties have been shipped for translation. -# ============================================================================== diff --git a/plugins/org.eclipse.net4j/plugin.xml b/plugins/org.eclipse.net4j/plugin.xml deleted file mode 100644 index 347e7300a1..0000000000 --- a/plugins/org.eclipse.net4j/plugin.xml +++ /dev/null @@ -1,3 +0,0 @@ -<?xml version="1.0" encoding="UTF-8"?> -<?eclipse version="3.0"?> -<plugin /> diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/bundle/Activator.java b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/bundle/Activator.java new file mode 100644 index 0000000000..fb5f2862f5 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/bundle/Activator.java @@ -0,0 +1,15 @@ +package org.eclipse.internal.net4j.util.bundle; + +import org.osgi.framework.BundleActivator; +import org.osgi.framework.BundleContext; + +public class Activator implements BundleActivator +{ + public void start(BundleContext context) throws Exception + { + } + + public void stop(BundleContext context) throws Exception + { + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/bundle/Log.java b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/bundle/Log.java new file mode 100644 index 0000000000..ade141c769 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/bundle/Log.java @@ -0,0 +1,113 @@ +package org.eclipse.internal.net4j.util.bundle; + +import org.osgi.framework.BundleContext; +import org.osgi.util.tracker.ServiceTracker; + + +public final class Log +{ + private static ServiceTracker logTracker; + + private Log() + { + }; + + static void init(BundleContext bc) + { + logTracker = new ServiceTracker(bc, LogService.class.getName(), null); + logTracker.open(); + } + + static void dispose() + { + if (logTracker != null) + { + logTracker.close(); + logTracker = null; + } + } + + public static void debug(String message) + { + log(LogService.LOG_DEBUG, message, null); + } + + public static void debug(String message, Throwable t) + { + log(LogService.LOG_DEBUG, message, null); + } + + public static void info(String message) + { + log(LogService.LOG_INFO, message, null); + } + + public static void info(String message, Throwable t) + { + log(LogService.LOG_INFO, message, null); + } + + public static void warn(String message) + { + log(LogService.LOG_WARNING, message, null); + } + + public static void warn(String message, Throwable t) + { + log(LogService.LOG_WARNING, message, null); + } + + public static void error(String message) + { + log(LogService.LOG_ERROR, message, null); + } + + public static void error(String message, Throwable t) + { + log(LogService.LOG_ERROR, message, null); + } + + public static void log(int level, String message) + { + log(level, message, null); + } + + public static void log(int level, String message, Throwable t) + { + LogService logService = (LogService)logTracker.getService(); + if (logService != null) + { + logService.log(level, message, t); + } + else + { + switch (level) + { + case LogService.LOG_DEBUG: + System.out.print("[DEBUG] "); + break; + + case LogService.LOG_INFO: + System.out.print("[INFO] "); + break; + + case LogService.LOG_WARNING: + System.out.print("[WARN] "); + break; + + case LogService.LOG_ERROR: + System.out.print("[ERROR] "); + break; + + default: + break; + } + + System.out.println(message); + if (t != null) + { + t.printStackTrace(); + } + } + } +}
\ No newline at end of file diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/AbstractConnector.java b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/AbstractConnector.java new file mode 100644 index 0000000000..f1dc579db4 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/AbstractConnector.java @@ -0,0 +1,567 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.internal.net4j.transport; + +import org.eclipse.net4j.transport.Buffer; +import org.eclipse.net4j.transport.BufferProvider; +import org.eclipse.net4j.transport.Channel; +import org.eclipse.net4j.transport.Connector; +import org.eclipse.net4j.transport.ConnectorCredentials; +import org.eclipse.net4j.transport.ConnectorException; +import org.eclipse.net4j.transport.Protocol; +import org.eclipse.net4j.transport.ProtocolFactory; +import org.eclipse.net4j.util.lifecycle.AbstractLifecycle; +import org.eclipse.net4j.util.lifecycle.LifecycleListener; +import org.eclipse.net4j.util.lifecycle.LifecycleNotifier; +import org.eclipse.net4j.util.lifecycle.LifecycleUtil; +import org.eclipse.net4j.util.registry.IRegistry; + +import java.util.ArrayList; +import java.util.List; +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; + +/** + * @author Eike Stepper + */ +public abstract class AbstractConnector extends AbstractLifecycle implements Connector, + BufferProvider +{ + private static final ChannelImpl NULL_CHANNEL = new ChannelImpl(null); + + private ConnectorCredentials credentials; + + private IRegistry<String, ProtocolFactory> protocolFactoryRegistry; + + private BufferProvider bufferProvider; + + /** + * An optional executor to be used by the {@link Channel}s to process their + * {@link ChannelImpl#receiveQueue} instead of the current thread. If not + * <code>null</code> the sender and the receiver peers become decoupled. + * <p> + */ + private ExecutorService receiveExecutor; + + /** + * TODO synchronize on channels? + */ + private List<ChannelImpl> channels = new ArrayList(); + + private State state = State.DISCONNECTED; + + /** + * Don't initialize lazily to circumvent synchronization! + */ + private Queue<StateListener> stateListeners = new ConcurrentLinkedQueue(); + + /** + * Don't initialize lazily to circumvent synchronization! + */ + private Queue<ChannelListener> channelListeners = new ConcurrentLinkedQueue(); + + /** + * Is registered with each {@link Channel} of this {@link Connector}. + * <p> + */ + private LifecycleListener channelLifecycleListener = new ChannelLifecycleListener(); + + private CountDownLatch finishedConnecting; + + private CountDownLatch finishedNegotiating; + + public AbstractConnector() + { + } + + public abstract void multiplexBuffer(Channel channel); + + public ExecutorService getReceiveExecutor() + { + return receiveExecutor; + } + + public void setReceiveExecutor(ExecutorService receiveExecutor) + { + this.receiveExecutor = receiveExecutor; + } + + public IRegistry<String, ProtocolFactory> getProtocolFactoryRegistry() + { + return protocolFactoryRegistry; + } + + public void setProtocolFactoryRegistry(IRegistry<String, ProtocolFactory> protocolFactoryRegistry) + { + this.protocolFactoryRegistry = protocolFactoryRegistry; + } + + public void addStateListener(StateListener listener) + { + stateListeners.add(listener); + } + + public void removeStateListener(StateListener listener) + { + stateListeners.remove(listener); + } + + public void addChannelListener(ChannelListener listener) + { + channelListeners.add(listener); + } + + public void removeChannelListener(ChannelListener listener) + { + channelListeners.remove(listener); + } + + public BufferProvider getBufferProvider() + { + return bufferProvider; + } + + public void setBufferProvider(BufferProvider bufferProvider) + { + this.bufferProvider = bufferProvider; + } + + public short getBufferCapacity() + { + return bufferProvider.getBufferCapacity(); + } + + public Buffer provideBuffer() + { + return bufferProvider.provideBuffer(); + } + + public void retainBuffer(Buffer buffer) + { + bufferProvider.retainBuffer(buffer); + } + + public boolean isClient() + { + return getType() == Type.CLIENT; + } + + public boolean isServer() + { + return getType() == Type.SERVER; + } + + public ConnectorCredentials getCredentials() + { + return credentials; + } + + public void setCredentials(ConnectorCredentials credentials) + { + this.credentials = credentials; + } + + public State getState() + { + return state; + } + + public void setState(State newState) throws ConnectorException + { + State oldState = getState(); + if (newState != oldState) + { + System.out.println(toString() + ": Setting state " + newState + " (was " + + oldState.toString().toLowerCase() + ")"); + state = newState; + fireStateChanged(newState, oldState); + + switch (newState) + { + case DISCONNECTED: + if (finishedConnecting != null) + { + finishedConnecting.countDown(); + finishedConnecting = null; + } + + if (finishedNegotiating != null) + { + finishedNegotiating.countDown(); + finishedNegotiating = null; + } + break; + + case CONNECTING: + finishedConnecting = new CountDownLatch(1); + finishedNegotiating = new CountDownLatch(1); + if (getType() == Type.SERVER) + { + setState(State.NEGOTIATING); + } + break; + + case NEGOTIATING: + finishedConnecting.countDown(); + setState(State.CONNECTED); // TODO Implement negotiation + break; + + case CONNECTED: + finishedConnecting.countDown(); // Just in case of suspicion + finishedNegotiating.countDown(); + break; + + } + } + } + + public boolean isConnected() + { + return getState() == State.CONNECTED; + } + + public void connectAsync() throws ConnectorException + { + try + { + activate(); + } + catch (ConnectorException ex) + { + throw ex; + } + catch (Exception ex) + { + throw new ConnectorException(ex); + } + } + + public boolean waitForConnection(long timeout) throws ConnectorException + { + State state = getState(); + if (state == State.DISCONNECTED) + { + return false; + } + + try + { + System.out.println(toString() + ": Waiting for connection..."); + return finishedNegotiating.await(timeout, TimeUnit.MILLISECONDS); + } + catch (InterruptedException ex) + { + return false; + } + } + + public boolean connect(long timeout) throws ConnectorException + { + connectAsync(); + return waitForConnection(timeout); + } + + public ConnectorException disconnect() + { + Exception ex = deactivate(); + if (ex == null) + { + return null; + } + + if (ex instanceof ConnectorException) + { + return (ConnectorException)ex; + } + + return new ConnectorException(ex); + } + + public Channel[] getChannels() + { + final List<Channel> result = new ArrayList<Channel>(); + synchronized (channels) + { + for (final ChannelImpl channel : channels) + { + if (channel != NULL_CHANNEL) + { + result.add(channel); + } + } + } + + return result.toArray(new Channel[result.size()]); + } + + public Channel openChannel() throws ConnectorException + { + return openChannel(null); + } + + public Channel openChannel(String protocolID) throws ConnectorException + { + short channelID = findFreeChannelID(); + ChannelImpl channel = createChannel(channelID, protocolID); + registerChannelWithPeer(channelID, protocolID); + + try + { + channel.activate(); + } + catch (ConnectorException ex) + { + throw ex; + } + catch (Exception ex) + { + throw new ConnectorException(ex); + } + + return channel; + } + + public ChannelImpl createChannel(short channelID, String protocolID) + { + ChannelImpl channel = new ChannelImpl(receiveExecutor); + Protocol protocol = createProtocol(protocolID, channel); + if (protocol == null) + { + System.out.println(toString() + ": Opening channel without protocol"); + } + else + { + System.out.println(toString() + ": Opening channel with protocol " + protocolID); + } + + channel.setChannelID(channelID); + channel.setConnector(this); + channel.setReceiveHandler(protocol); + channel.addLifecycleListener(channelLifecycleListener); + addChannel(channel); + return channel; + } + + public ChannelImpl getChannel(short channelID) + { + try + { + ChannelImpl channel = channels.get(channelID); + if (channel == null || channel == NULL_CHANNEL) + { + throw new NullPointerException(); + } + + return channel; + } + catch (IndexOutOfBoundsException ex) + { + System.out.println(toString() + ": Invalid channelID " + channelID); + return null; + } + } + + protected List<Queue<Buffer>> getChannelBufferQueues() + { + final List<Queue<Buffer>> result = new ArrayList(); + synchronized (channels) + { + for (final ChannelImpl channel : channels) + { + if (channel != NULL_CHANNEL) + { + Queue<Buffer> bufferQueue = channel.getSendQueue(); + result.add(bufferQueue); + } + } + } + + return result; + } + + protected short findFreeChannelID() + { + synchronized (channels) + { + int size = channels.size(); + for (short i = 0; i < size; i++) + { + if (channels.get(i) == NULL_CHANNEL) + { + return i; + } + } + + channels.add(NULL_CHANNEL); + return (short)size; + } + } + + protected void addChannel(ChannelImpl channel) + { + short channelID = channel.getChannelID(); + while (channelID >= channels.size()) + { + channels.add(NULL_CHANNEL); + } + + channels.set(channelID, channel); + } + + protected void removeChannel(ChannelImpl channel) + { + channel.removeLifecycleListener(channelLifecycleListener); + int channelID = channel.getChannelID(); + + System.out.println(toString() + ": Removing channel " + channelID); + channels.set(channelID, NULL_CHANNEL); + } + + protected Protocol createProtocol(String protocolID, Channel channel) + { + if (protocolID == null || protocolID.length() == 0) + { + return null; + } + + IRegistry<String, ProtocolFactory> registry = getProtocolFactoryRegistry(); + if (registry == null) + { + return null; + } + + ProtocolFactory factory = registry.lookup(protocolID); + if (factory == null) + { + System.out.println(toString() + ": Unknown protocol " + protocolID); + return null; + } + + return factory.createProtocol(channel); + } + + protected void fireChannelOpened(Channel channel) + { + for (ChannelListener listener : channelListeners) + { + try + { + listener.notifyChannelOpened(channel); + } + catch (Exception ex) + { + ex.printStackTrace(); + } + } + } + + protected void fireChannelClosing(Channel channel) + { + for (ChannelListener listener : channelListeners) + { + try + { + listener.notifyChannelClosing(channel); + } + catch (Exception ex) + { + ex.printStackTrace(); + } + } + } + + protected void fireStateChanged(State newState, State oldState) + { + for (StateListener listener : stateListeners) + { + try + { + listener.notifyStateChanged(this, newState, oldState); + } + catch (Exception ex) + { + ex.printStackTrace(); + } + } + } + + @Override + protected void onAccessBeforeActivate() throws Exception + { + super.onAccessBeforeActivate(); + if (bufferProvider == null) + { + throw new IllegalStateException("bufferProvider == null"); + } + + if (protocolFactoryRegistry == null) + { + System.out.println(toString() + ": (INFO) protocolFactoryRegistry == null"); + } + + if (receiveExecutor == null) + { + System.out.println(toString() + ": (INFO) receiveExecutor == null"); + } + } + + @Override + protected void onActivate() throws Exception + { + super.onActivate(); + setState(State.CONNECTING); + } + + @Override + protected void onDeactivate() throws Exception + { + setState(State.DISCONNECTED); + for (short i = 0; i < channels.size(); i++) + { + ChannelImpl channel = channels.get(i); + if (channel != null) + { + LifecycleUtil.deactivate(channel); + } + } + + channels.clear(); + super.onDeactivate(); + } + + protected abstract void registerChannelWithPeer(short channelID, String protocolID) + throws ConnectorException; + + /** + * Is registered with each {@link Channel} of this {@link Connector}. + * <p> + * + * @author Eike Stepper + */ + private final class ChannelLifecycleListener implements LifecycleListener + { + public void notifyLifecycleActivated(LifecycleNotifier notifier) + { + ChannelImpl channel = (ChannelImpl)notifier; + fireChannelOpened(channel); + } + + public void notifyLifecycleDeactivating(LifecycleNotifier notifier) + { + ChannelImpl channel = (ChannelImpl)notifier; + fireChannelClosing(channel); + removeChannel(channel); + } + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/AbstractProtocol.java b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/AbstractProtocol.java new file mode 100644 index 0000000000..dabe8e146e --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/AbstractProtocol.java @@ -0,0 +1,54 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.internal.net4j.transport; + +import org.eclipse.net4j.transport.Buffer; +import org.eclipse.net4j.transport.BufferProvider; +import org.eclipse.net4j.transport.Channel; +import org.eclipse.net4j.transport.Protocol; + +/** + * @author Eike Stepper + */ +public abstract class AbstractProtocol implements Protocol, BufferProvider +{ + private Channel channel; + + public AbstractProtocol(Channel channel) + { + this.channel = channel; + } + + public Channel getChannel() + { + return channel; + } + + public short getBufferCapacity() + { + return BufferUtil.getBufferProvider(channel).getBufferCapacity(); + } + + public Buffer provideBuffer() + { + return BufferUtil.getBufferProvider(channel).provideBuffer(); + } + + public void retainBuffer(Buffer buffer) + { + BufferUtil.getBufferProvider(channel).retainBuffer(buffer); + } + + public void dispose() + { + channel = null; + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/AbstractProtocolFactory.java b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/AbstractProtocolFactory.java new file mode 100644 index 0000000000..ccd55fe90a --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/AbstractProtocolFactory.java @@ -0,0 +1,39 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.internal.net4j.transport; + +import org.eclipse.net4j.transport.ProtocolFactory; +import org.eclipse.net4j.transport.Connector.Type; + +/** + * @author Eike Stepper + */ +public abstract class AbstractProtocolFactory implements ProtocolFactory +{ + public final boolean isForClients() + { + return getConnectorTypes().contains(Type.CLIENT); + } + + public final boolean isForServers() + { + return getConnectorTypes().contains(Type.SERVER); + } + + public final boolean isSymmetric() + { + return isForClients() && isForServers(); + } + + public void dispose() + { + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/BufferFactoryImpl.java b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/BufferFactoryImpl.java new file mode 100644 index 0000000000..43ed786cd4 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/BufferFactoryImpl.java @@ -0,0 +1,48 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.internal.net4j.transport; + +import org.eclipse.net4j.transport.Buffer; + +/** + * @author Eike Stepper + */ +public class BufferFactoryImpl extends BufferProviderImpl +{ + public BufferFactoryImpl(short bufferCapacity) + { + super(bufferCapacity); + } + + @Override + protected Buffer doProvideBuffer() + { + System.out.println(toString() + ": Creating buffer of capacity " + getBufferCapacity()); + return new BufferImpl(this, getBufferCapacity()); + } + + @Override + protected void doRetainBuffer(Buffer buffer) + { + if (buffer instanceof BufferImpl) + { + ((BufferImpl)buffer).dispose(); + } + + buffer = null; + } + + @Override + public String toString() + { + return "BufferFactory[capacity=" + getBufferCapacity() + "]"; + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/BufferImpl.java b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/BufferImpl.java new file mode 100644 index 0000000000..aa3e2ed908 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/BufferImpl.java @@ -0,0 +1,254 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.internal.net4j.transport; + +import org.eclipse.net4j.transport.Buffer; +import org.eclipse.net4j.transport.BufferProvider; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.channels.ClosedChannelException; +import java.nio.channels.SocketChannel; + +/** + * @author Eike Stepper + */ +public class BufferImpl implements Buffer +{ + public static final short HEADER_SIZE = 4; + + public static final short NO_CHANNEL = Short.MIN_VALUE; + + private static final int EOS_OFFSET = 1; + + private BufferProvider bufferProvider; + + private short channelID; + + private boolean eos; + + private State state = State.INITIAL; + + private ByteBuffer byteBuffer; + + public BufferImpl(BufferProvider provider, short capacity) + { + this.bufferProvider = provider; + byteBuffer = ByteBuffer.allocateDirect(capacity); + } + + public boolean isEOS() + { + return eos; + } + + public void setEOS(boolean eos) + { + this.eos = eos; + } + + public BufferProvider getBufferProvider() + { + return bufferProvider; + } + + public void setBufferProvider(BufferProvider bufferProvider) + { + this.bufferProvider = bufferProvider; + } + + public short getChannelID() + { + if (state == State.INITIAL || state == State.READING_HEADER) + { + throw new IllegalStateException("state == " + state); + } + + return channelID; + } + + public short getCapacity() + { + return (short)byteBuffer.capacity(); + } + + public ByteBuffer getByteBuffer() + { + if (state != State.GETTING && state != State.PUTTING) + { + throw new IllegalStateException("state == " + state); + } + + return byteBuffer; + } + + public State getState() + { + return state; + } + + /** + * TODO Check for multiply released buffers? + */ + public void release() + { + if (bufferProvider != null) + { + bufferProvider.retainBuffer(this); + } + } + + public void clear() + { + byteBuffer.clear(); + state = State.INITIAL; + channelID = NO_CHANNEL; + eos = false; + } + + public void dispose() + { + bufferProvider = null; + byteBuffer = null; + } + + public ByteBuffer startGetting(SocketChannel socketChannel) throws IOException + { + if (state != State.INITIAL && state != State.READING_HEADER && state != State.READING_BODY) + { + throw new IllegalStateException("state == " + state); + } + + if (state == State.INITIAL) + { + byteBuffer.limit(BufferImpl.HEADER_SIZE); + state = State.READING_HEADER; + } + + if (state == State.READING_HEADER) + { + int num = socketChannel.read(byteBuffer); + if (num == -1) + { + throw new ClosedChannelException(); + } + + if (byteBuffer.hasRemaining()) + { + return null; + } + + byteBuffer.flip(); + channelID = byteBuffer.getShort(); + short payloadSize = byteBuffer.getShort(); + if (payloadSize < 0) + { + eos = true; + payloadSize = (short)-payloadSize; + } + + payloadSize -= EOS_OFFSET; + System.out.println(toString() + ": Read " + (HEADER_SIZE + payloadSize) + " bytes" + + (eos ? " (EOS)" : "")); + + byteBuffer.clear(); + byteBuffer.limit(payloadSize); + state = State.READING_BODY; + } + + // state == State.READING_BODY + if (socketChannel.read(byteBuffer) == -1) + { + throw new ClosedChannelException(); + } + + if (byteBuffer.hasRemaining()) + { + return null; + } + + byteBuffer.flip(); + state = State.GETTING; + return byteBuffer; + } + + public ByteBuffer startPutting(short channelID) + { + if (state != State.INITIAL) + { + throw new IllegalStateException("state == " + state); + } + + state = State.PUTTING; + this.channelID = channelID; + + byteBuffer.clear(); + byteBuffer.position(BufferImpl.HEADER_SIZE); + return byteBuffer; + } + + public boolean write(SocketChannel socketChannel) throws IOException + { + if (state != State.PUTTING && state != State.WRITING) + { + throw new IllegalStateException("state == " + state); + } + + if (state != State.WRITING) + { + if (channelID == NO_CHANNEL) + { + throw new IllegalStateException("channelID == NO_CHANNEL"); + } + + int payloadSize = byteBuffer.position() - BufferImpl.HEADER_SIZE + EOS_OFFSET; + if (eos) + { + payloadSize = -payloadSize; + } + + byteBuffer.flip(); + byteBuffer.putShort(channelID); + byteBuffer.putShort((short)payloadSize); + byteBuffer.position(0); + state = State.WRITING; + } + + if (socketChannel.write(byteBuffer) == -1) + { + throw new IOException("Channel closed"); + } + + if (byteBuffer.hasRemaining()) + { + return false; + } + + System.out.println(toString() + ": Wrote " + byteBuffer.limit() + " bytes" + + (eos ? " (EOS)" : "")); + clear(); + return true; + } + + @Override + public String toString() + { + return "Buffer[channelID=" + channelID + ", state=" + state + "]"; + } + + /** + * @author Eike Stepper + */ + public static enum State + { + INITIAL, PUTTING, WRITING, READING_HEADER, READING_BODY, GETTING + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/BufferPoolImpl.java b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/BufferPoolImpl.java new file mode 100644 index 0000000000..3f3b254b9a --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/BufferPoolImpl.java @@ -0,0 +1,110 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.internal.net4j.transport; + +import org.eclipse.net4j.transport.Buffer; +import org.eclipse.net4j.transport.BufferPool; +import org.eclipse.net4j.transport.BufferProvider; + +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; + +/** + * @author Eike Stepper + */ +public class BufferPoolImpl extends BufferProviderImpl implements BufferPool, + BufferPool.Introspection +{ + private final BufferProvider factory; + + private final Queue<Buffer> queue = new ConcurrentLinkedQueue<Buffer>(); + + private int pooledBuffers; + + public BufferPoolImpl(BufferProvider factory) + { + super(factory.getBufferCapacity()); + this.factory = factory; + } + + public int getPooledBuffers() + { + return pooledBuffers; + } + + public boolean evictOne() + { + Buffer buffer = queue.poll(); + if (buffer == null) + { + return false; + } + + System.out.println(toString() + ": Evicting buffer"); + factory.retainBuffer(buffer); + --pooledBuffers; + return true; + } + + public int evict(int survivors) + { + int evictedBuffers = 0; + while (pooledBuffers > survivors) + { + if (evictOne()) + { + ++evictedBuffers; + } + else + { + break; + } + } + + return evictedBuffers; + } + + @Override + protected Buffer doProvideBuffer() + { + Buffer buffer = queue.poll(); + if (buffer != null) + { + System.out.println(toString() + ": Obtaining buffer"); + } + else + { + buffer = factory.provideBuffer(); + ((BufferImpl)buffer).setBufferProvider(this); + } + + buffer.clear(); + return buffer; + } + + @Override + protected void doRetainBuffer(Buffer buffer) + { + if (buffer.getCapacity() != getBufferCapacity()) + { + throw new IllegalArgumentException("buffer.getCapacity() != getBufferCapacity()"); + } + + System.out.println(toString() + ": Retaining buffer"); + queue.add(buffer); + } + + @Override + public String toString() + { + return "BufferPool[size=" + pooledBuffers + ", " + factory + "]"; + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/BufferProviderImpl.java b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/BufferProviderImpl.java new file mode 100644 index 0000000000..d6ba2d9526 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/BufferProviderImpl.java @@ -0,0 +1,68 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.internal.net4j.transport; + +import org.eclipse.net4j.transport.Buffer; +import org.eclipse.net4j.transport.BufferProvider; + +/** + * @author Eike Stepper + */ +public abstract class BufferProviderImpl implements BufferProvider, BufferProvider.Introspection +{ + private short bufferCapacity; + + private long providedBuffers; + + private long retainedBuffers; + + public BufferProviderImpl(short bufferCapacity) + { + this.bufferCapacity = bufferCapacity; + } + + public final long getProvidedBuffers() + { + return providedBuffers; + } + + public final long getRetainedBuffers() + { + return retainedBuffers; + } + + public final short getBufferCapacity() + { + return bufferCapacity; + } + + public final Buffer provideBuffer() + { + ++providedBuffers; + return doProvideBuffer(); + } + + public final void retainBuffer(Buffer buffer) + { + ++retainedBuffers; + doRetainBuffer(buffer); + } + + @Override + public String toString() + { + return "BufferProvider[capacity=" + bufferCapacity + "]"; + } + + protected abstract Buffer doProvideBuffer(); + + protected abstract void doRetainBuffer(Buffer buffer); +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/BufferUtil.java b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/BufferUtil.java new file mode 100644 index 0000000000..b8847df93a --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/BufferUtil.java @@ -0,0 +1,140 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.internal.net4j.transport; + +import org.eclipse.net4j.transport.BufferProvider; + +import java.io.UnsupportedEncodingException; +import java.nio.ByteBuffer; + +/** + * @author Eike Stepper + */ +public final class BufferUtil +{ + public static final short DEFAULT_BUFFER_CAPACITY = 4096; + + public static final String UTF8_CHAR_SET_NAME = "UTF-8"; + + private BufferUtil() + { + } + + public static BufferProvider createBufferFactory(short bufferCapacity) + { + return new BufferFactoryImpl(bufferCapacity); + } + + public static BufferProvider createBufferFactory() + { + return new BufferFactoryImpl(DEFAULT_BUFFER_CAPACITY); + } + + public static BufferProvider createBufferPool(BufferProvider factory) + { + return new BufferPoolImpl(factory); + } + + public static BufferProvider createBufferPool(short bufferCapacity) + { + return createBufferPool(createBufferFactory(bufferCapacity)); + } + + public static BufferProvider createBufferPool() + { + return createBufferPool(DEFAULT_BUFFER_CAPACITY); + } + + public static BufferProvider getBufferProvider(Object object) + { + if (object instanceof BufferProvider) + { + return (BufferProvider)object; + } + + if (object == null) + { + throw new IllegalArgumentException("object == null"); + } + + throw new IllegalArgumentException("Unable to provide buffers: " + object); + } + + public static byte[] toUTF8(String str) + { + if (str == null) + { + return new byte[0]; + } + + try + { + byte[] bytes = str.getBytes(UTF8_CHAR_SET_NAME); + String test = new String(bytes, UTF8_CHAR_SET_NAME); + if (!str.equals(test)) + { + throw new IllegalArgumentException("String not encodable: " + str); + } + + return bytes; + } + catch (UnsupportedEncodingException ex) + { + // This should really not happen + throw new RuntimeException(ex); + } + } + + public static String fromUTF8(byte[] bytes) + { + try + { + return new String(bytes, UTF8_CHAR_SET_NAME); + } + catch (UnsupportedEncodingException ex) + { + // This should really not happen + throw new RuntimeException(ex); + } + } + + public static void putByteArray(ByteBuffer byteBuffer, byte[] array) + { + byteBuffer.putShort((short)array.length); + if (array.length != 0) + { + byteBuffer.put(array); + } + } + + public static byte[] getByteArray(ByteBuffer byteBuffer) + { + short length = byteBuffer.getShort(); + byte[] array = new byte[length]; + if (length != 0) + { + byteBuffer.get(array); + } + + return array; + } + + public static void putUTF8(ByteBuffer byteBuffer, String str) + { + byte[] bytes = BufferUtil.toUTF8(str); + if (bytes.length > byteBuffer.remaining()) + { + throw new IllegalArgumentException("String too long: " + str); + } + + putByteArray(byteBuffer, bytes); + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/ChannelImpl.java b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/ChannelImpl.java new file mode 100644 index 0000000000..f4cce359d2 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/ChannelImpl.java @@ -0,0 +1,247 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.internal.net4j.transport; + +import org.eclipse.net4j.transport.Buffer; +import org.eclipse.net4j.transport.BufferHandler; +import org.eclipse.net4j.transport.BufferProvider; +import org.eclipse.net4j.transport.Channel; +import org.eclipse.net4j.transport.Connector; +import org.eclipse.net4j.util.lifecycle.AbstractLifecycle; + +import org.eclipse.internal.net4j.transport.BufferImpl.State; + +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.ExecutorService; + +/** + * @author Eike Stepper + */ +public class ChannelImpl extends AbstractLifecycle implements Channel, BufferProvider, Runnable +{ + private short channelID = BufferImpl.NO_CHANNEL; + + private AbstractConnector connector; + + /** + * The external handler for buffers passed from the {@link #connector}. + * <p> + */ + private BufferHandler receiveHandler; + + /** + * An optional executor that is used to process the {@link #receiveQueue} + * instead of the current thread. If not <code>null</code> the sender and + * the receiver peers become decoupled. + * <p> + */ + private ExecutorService receiveExecutor; + + private Occupation receiveExecutorOccupation = new Occupation(); + + /** + * TODO Optimize for embedded transport + */ + private Queue<Buffer> receiveQueue = new ConcurrentLinkedQueue(); + + private Queue<Buffer> sendQueue = new ConcurrentLinkedQueue(); + + public ChannelImpl(ExecutorService receiveExecutor) + { + this.receiveExecutor = receiveExecutor; + } + + public short getChannelID() + { + return channelID; + } + + public void setChannelID(short channelID) + { + if (channelID == BufferImpl.NO_CHANNEL) + { + throw new IllegalArgumentException("channelID == INVALID_CHANNEL_ID"); + } + + this.channelID = channelID; + } + + public Connector getConnector() + { + return connector; + } + + public void setConnector(AbstractConnector connector) + { + this.connector = connector; + } + + public short getBufferCapacity() + { + return BufferUtil.getBufferProvider(connector).getBufferCapacity(); + } + + public Buffer provideBuffer() + { + return BufferUtil.getBufferProvider(connector).provideBuffer(); + } + + public void retainBuffer(Buffer buffer) + { + BufferUtil.getBufferProvider(connector).retainBuffer(buffer); + } + + public Queue<Buffer> getSendQueue() + { + return sendQueue; + } + + public ExecutorService getReceiveExecutor() + { + return receiveExecutor; + } + + public Queue<Buffer> getReceiveQueue() + { + return receiveQueue; + } + + public BufferHandler getReceiveHandler() + { + return receiveHandler; + } + + public void setReceiveHandler(BufferHandler receiveHandler) + { + this.receiveHandler = receiveHandler; + } + + public void close() + { + deactivate(); + } + + public void sendBuffer(Buffer buffer) + { + handleBuffer(buffer); + } + + public void handleBuffer(Buffer buffer) + { + State state = ((BufferImpl)buffer).getState(); + if (state != State.PUTTING) + { + System.out.println(toString() + ": Ignoring buffer in state == " + state); + return; + } + + sendQueue.add(buffer); + connector.multiplexBuffer(this); + } + + public void handleBufferFromMultiplexer(Buffer buffer) + { + if (receiveHandler == null) + { + System.out.println(toString() + ": Ignoring buffer because receiveHandler == null"); + buffer.release(); + return; + } + + if (receiveExecutor == null) + { + // Bypass the receiveQueue + receiveHandler.handleBuffer(buffer); + return; + } + + receiveQueue.add(buffer); + + // isOccupied can (and must) be called unsynchronized here + if (receiveExecutorOccupation.isOccupied()) + { + return; + } + + synchronized (receiveExecutorOccupation) + { + receiveExecutorOccupation.setOccupied(true); + } + + System.out.println(toString() + ": Spawning new receive executor"); + receiveExecutor.execute(this); + } + + /** + * Executed in the context of the {@link #receiveExecutor}. + * <p> + */ + public void run() + { + synchronized (receiveExecutorOccupation) + { + Buffer buffer; + while ((buffer = receiveQueue.poll()) != null) + { + receiveHandler.handleBuffer(buffer); + } + + receiveExecutorOccupation.setOccupied(false); + } + } + + @Override + public String toString() + { + return "Channel[" + connector + ":" + channelID + "]"; + } + + @Override + protected void onAccessBeforeActivate() throws Exception + { + super.onAccessBeforeActivate(); + if (channelID == BufferImpl.NO_CHANNEL) + { + throw new IllegalStateException("channelID == INVALID_CHANNEL_ID"); + } + + if (connector == null) + { + throw new IllegalStateException("connector == null"); + } + } + + @Override + protected void onDeactivate() throws Exception + { + sendQueue.clear(); + super.onDeactivate(); + } + + /** + * @author Eike Stepper + */ + private static class Occupation + { + private boolean occupied = false; + + public boolean isOccupied() + { + return occupied; + } + + public void setOccupied(boolean occupied) + { + this.occupied = occupied; + } + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/ConnectorCredentialsImpl.java b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/ConnectorCredentialsImpl.java new file mode 100644 index 0000000000..a9d87dbfdf --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/ConnectorCredentialsImpl.java @@ -0,0 +1,31 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.internal.net4j.transport; + +import org.eclipse.net4j.transport.ConnectorCredentials; + +/** + * @author Eike Stepper + */ +public class ConnectorCredentialsImpl implements ConnectorCredentials +{ + private String userID; + + public ConnectorCredentialsImpl(String userID) + { + this.userID = userID; + } + + public String getUserID() + { + return userID; + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/embedded/AbstractEmbeddedConnector.java b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/embedded/AbstractEmbeddedConnector.java new file mode 100644 index 0000000000..644dde6875 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/embedded/AbstractEmbeddedConnector.java @@ -0,0 +1,90 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.internal.net4j.transport.embedded; + +import org.eclipse.net4j.transport.Buffer; +import org.eclipse.net4j.transport.Channel; +import org.eclipse.net4j.transport.ConnectorException; +import org.eclipse.net4j.util.lifecycle.LifecycleUtil; + +import org.eclipse.internal.net4j.transport.AbstractConnector; +import org.eclipse.internal.net4j.transport.ChannelImpl; + +import java.util.Queue; + +/** + * TODO Remove peer channels + * + * @author Eike Stepper + */ +public abstract class AbstractEmbeddedConnector extends AbstractConnector +{ + private AbstractEmbeddedConnector peer; + + public AbstractEmbeddedConnector() + { + } + + public AbstractEmbeddedConnector getPeer() + { + return peer; + } + + public void setPeer(AbstractEmbeddedConnector peer) + { + this.peer = peer; + } + + @Override + protected void registerChannelWithPeer(short channelID, String protocolID) + throws ConnectorException + { + try + { + ChannelImpl channel = getPeer().createChannel(channelID, protocolID); + if (channel == null) + { + throw new ConnectorException("Failed to register channel with peer"); + } + + channel.activate(); + } + catch (ConnectorException ex) + { + throw ex; + } + catch (Exception ex) + { + throw new ConnectorException(ex); + } + } + + public void multiplexBuffer(Channel localChannel) + { + short channelID = localChannel.getChannelID(); + ChannelImpl peerChannel = peer.getChannel(channelID); + if (peerChannel == null) + { + throw new IllegalStateException("peerChannel == null"); + } + + Queue<Buffer> localQueue = ((ChannelImpl)localChannel).getSendQueue(); + Buffer buffer = localQueue.poll(); + peerChannel.handleBufferFromMultiplexer(buffer); + } + + @Override + protected void onDeactivate() throws Exception + { + LifecycleUtil.deactivateNoisy(getPeer()); + super.onDeactivate(); + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/embedded/ClientEmbeddedConnectorImpl.java b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/embedded/ClientEmbeddedConnectorImpl.java new file mode 100644 index 0000000000..7acc06d182 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/embedded/ClientEmbeddedConnectorImpl.java @@ -0,0 +1,34 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.internal.net4j.transport.embedded; + +/** + * @author Eike Stepper + */ +public class ClientEmbeddedConnectorImpl extends AbstractEmbeddedConnector +{ + public Type getType() + { + return Type.CLIENT; + } + + @Override + protected void onActivate() throws Exception + { + super.onActivate(); + setPeer(createServerPeer()); + } + + protected AbstractEmbeddedConnector createServerPeer() + { + return new ServerEmbeddedConnectorImpl(this); + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/embedded/EmbeddedUtil.java b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/embedded/EmbeddedUtil.java new file mode 100644 index 0000000000..182ca8e635 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/embedded/EmbeddedUtil.java @@ -0,0 +1,33 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.internal.net4j.transport.embedded; + +import org.eclipse.net4j.transport.BufferProvider; +import org.eclipse.net4j.transport.Connector; +import org.eclipse.net4j.transport.tcp.TCPSelector; + +/** + * @author Eike Stepper + */ +public final class EmbeddedUtil +{ + private EmbeddedUtil() + { + } + + public static Connector createConnector(BufferProvider bufferProvider, TCPSelector selector, + String host, int port) + { + ClientEmbeddedConnectorImpl connector = new ClientEmbeddedConnectorImpl(); + connector.setBufferProvider(bufferProvider); + return connector; + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/embedded/ServerEmbeddedConnectorImpl.java b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/embedded/ServerEmbeddedConnectorImpl.java new file mode 100644 index 0000000000..0e977588fa --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/embedded/ServerEmbeddedConnectorImpl.java @@ -0,0 +1,27 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.internal.net4j.transport.embedded; + +/** + * @author Eike Stepper + */ +public class ServerEmbeddedConnectorImpl extends AbstractEmbeddedConnector +{ + public ServerEmbeddedConnectorImpl(ClientEmbeddedConnectorImpl clientPeer) + { + setPeer(clientPeer); + } + + public Type getType() + { + return Type.SERVER; + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/tcp/AbstractTCPConnector.java b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/tcp/AbstractTCPConnector.java new file mode 100644 index 0000000000..3724e0025e --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/tcp/AbstractTCPConnector.java @@ -0,0 +1,352 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.internal.net4j.transport.tcp; + +import org.eclipse.net4j.transport.Buffer; +import org.eclipse.net4j.transport.Channel; +import org.eclipse.net4j.transport.ConnectorException; +import org.eclipse.net4j.transport.tcp.TCPConnector; +import org.eclipse.net4j.transport.tcp.TCPSelector; +import org.eclipse.net4j.transport.tcp.TCPSelectorListener; +import org.eclipse.net4j.util.lifecycle.LifecycleUtil; + +import org.eclipse.internal.net4j.transport.AbstractConnector; +import org.eclipse.internal.net4j.transport.ChannelImpl; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.channels.ClosedChannelException; +import java.nio.channels.SelectionKey; +import java.nio.channels.SocketChannel; +import java.util.List; +import java.util.Queue; + +/** + * @author Eike Stepper + */ +public abstract class AbstractTCPConnector extends AbstractConnector implements TCPConnector, + TCPSelectorListener.Active +{ + private SocketChannel socketChannel; + + private TCPSelector selector; + + private SelectionKey selectionKey; + + private Buffer inputBuffer; + + private ControlChannelImpl controlChannel; + + public AbstractTCPConnector() + { + try + { + socketChannel = SocketChannel.open(); + socketChannel.configureBlocking(false); + } + catch (IOException ex) + { + ex.printStackTrace(); + } + } + + /** + * SocketChannel must already be non-blocking! + */ + public AbstractTCPConnector(SocketChannel socketChannel) + { + this.socketChannel = socketChannel; + } + + public TCPSelector getSelector() + { + return selector; + } + + public void setSelector(TCPSelector selector) + { + this.selector = selector; + } + + public SocketChannel getSocketChannel() + { + return socketChannel; + } + + public SelectionKey getSelectionKey() + { + return selectionKey; + } + + /** + * Called by {@link ChannelImpl} each time a new buffer is available for + * multiplexing. This or another buffer can be dequeued from the outputQueue + * of the {@link ChannelImpl}. + */ + public void multiplexBuffer(Channel channel) + { + TCPUtil.setWriteInterest(selectionKey, true); + } + + public void handleConnect(TCPSelector selector, SocketChannel channel) + { + try + { + if (!channel.finishConnect()) + { + return; + } + } + catch (Exception ex) + { + return; + } + + try + { + TCPUtil.setConnectInterest(selectionKey, false); + TCPUtil.setReadInterest(selectionKey, true); + setState(State.NEGOTIATING); + } + catch (Exception ex) + { + ex.printStackTrace(); + deactivate(); + } + } + + public void handleRead(TCPSelector selector, SocketChannel socketChannel) + { + // TODO Is this needed? + if (!socketChannel.isConnected()) + { + deactivate(); + return; + } + + try + { + if (inputBuffer == null) + { + inputBuffer = getBufferProvider().provideBuffer(); + } + + ByteBuffer byteBuffer = inputBuffer.startGetting(socketChannel); + if (byteBuffer != null) + { + short channelID = inputBuffer.getChannelID(); + ChannelImpl channel = channelID == ControlChannelImpl.CONTROL_CHANNEL_ID ? controlChannel + : getChannel(channelID); + if (channel != null) + { + channel.handleBufferFromMultiplexer(inputBuffer); + } + else + { + System.out.println(toString() + ": Discarding buffer from unknown channel"); + inputBuffer.release(); + } + + inputBuffer = null; + } + } + catch (ClosedChannelException ex) + { + deactivate(); + } + catch (Exception ex) + { + ex.printStackTrace(); + deactivate(); + } + } + + public void handleWrite(TCPSelector selector, SocketChannel socketChannel) + { + // TODO Is this needed? + if (!socketChannel.isConnected()) + { + deactivate(); + return; + } + + try + { + boolean moreToWrite = false; + for (Queue<Buffer> bufferQueue : getChannelBufferQueues()) + { + Buffer buffer = bufferQueue.peek(); + if (buffer != null) + { + if (buffer.write(socketChannel)) + { + bufferQueue.poll(); + buffer.release(); + + if (!moreToWrite) + { + moreToWrite = !bufferQueue.isEmpty(); + } + } + else + { + moreToWrite = true; + break; + } + } + } + + if (!moreToWrite) + { + TCPUtil.setWriteInterest(selectionKey, false); + } + } + catch (ClosedChannelException ex) + { + deactivate(); + } + catch (Exception ex) + { + ex.printStackTrace(); + deactivate(); + } + } + + @Override + protected List<Queue<Buffer>> getChannelBufferQueues() + { + List<Queue<Buffer>> queues = super.getChannelBufferQueues(); + Queue<Buffer> controlQueue = controlChannel.getSendQueue(); + if (!controlQueue.isEmpty()) + { + queues.add(controlQueue); + } + + return queues; + } + + @Override + protected void registerChannelWithPeer(short channelID, String protocolID) + throws ConnectorException + { + if (!controlChannel.registerChannel(channelID, protocolID)) + { + throw new ConnectorException("Failed to register channel with peer"); + } + } + + @Override + protected void removeChannel(ChannelImpl channel) + { + if (isConnected()) + { + controlChannel.deregisterChannel(channel.getChannelID()); + } + + super.removeChannel(channel); + } + + @Override + protected void onAccessBeforeActivate() throws Exception + { + super.onAccessBeforeActivate(); + if (selector == null) + { + selector = TCPUtil.createTCPSelector(); + LifecycleUtil.activate(selector); + } + } + + @Override + protected void onActivate() throws Exception + { + super.onActivate(); + controlChannel = new ControlChannelImpl(this); + controlChannel.activate(); + + selectionKey = selector.register(socketChannel, this); + if (getType() == Type.SERVER) + { + TCPUtil.setConnectInterest(selectionKey, false); + } + } + + @Override + protected void onDeactivate() throws Exception + { + Exception exception = null; + + try + { + selectionKey.cancel(); + } + catch (RuntimeException ex) + { + if (exception == null) + { + exception = ex; + } + } + finally + { + selectionKey = null; + } + + try + { + socketChannel.close(); + } + catch (Exception ex) + { + if (exception == null) + { + exception = ex; + } + } + finally + { + socketChannel = null; + } + + try + { + controlChannel.deactivate(); + } + catch (Exception ex) + { + if (exception == null) + { + exception = ex; + } + } + finally + { + controlChannel = null; + } + + try + { + super.onDeactivate(); + } + catch (Exception ex) + { + if (exception == null) + { + exception = ex; + } + } + + if (exception != null) + { + throw exception; + } + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/tcp/ClientTCPConnectorImpl.java b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/tcp/ClientTCPConnectorImpl.java new file mode 100644 index 0000000000..769ab52787 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/tcp/ClientTCPConnectorImpl.java @@ -0,0 +1,82 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.internal.net4j.transport.tcp; + +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.nio.channels.SelectionKey; + +/** + * @author Eike Stepper + */ +public class ClientTCPConnectorImpl extends AbstractTCPConnector +{ + private String host; + + private int port = DEFAULT_PORT; + + public ClientTCPConnectorImpl() + { + } + + public String getHost() + { + return host; + } + + public void setHost(String host) + { + this.host = host; + } + + public int getPort() + { + return port; + } + + public void setPort(int port) + { + this.port = port; + } + + public Type getType() + { + return Type.CLIENT; + } + + @Override + public String toString() + { + return "ClientTCPConnector[" + host + ":" + port + "]"; + } + + @Override + protected void onAccessBeforeActivate() throws Exception + { + super.onAccessBeforeActivate(); + if (host == null || host.length() == 0) + { + throw new IllegalArgumentException("host == null || host.length() == 0"); + } + } + + @Override + protected void onActivate() throws Exception + { + super.onActivate(); + SelectionKey selKey = getSelectionKey(); + selKey.interestOps(selKey.interestOps() | SelectionKey.OP_CONNECT); + + InetAddress addr = InetAddress.getByName(host); + InetSocketAddress sAddr = new InetSocketAddress(addr, port); + getSocketChannel().connect(sAddr); + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/tcp/ControlChannelImpl.java b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/tcp/ControlChannelImpl.java new file mode 100644 index 0000000000..9a9e920d85 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/tcp/ControlChannelImpl.java @@ -0,0 +1,152 @@ +package org.eclipse.internal.net4j.transport.tcp; + +import org.eclipse.net4j.transport.Buffer; +import org.eclipse.net4j.util.concurrent.Synchronizer; +import org.eclipse.net4j.util.concurrent.SynchronizingCorrelator; + +import org.eclipse.internal.net4j.transport.BufferUtil; +import org.eclipse.internal.net4j.transport.ChannelImpl; + +import java.nio.ByteBuffer; + +/** + * @author Eike Stepper + */ +public final class ControlChannelImpl extends ChannelImpl +{ + public static final short CONTROL_CHANNEL_ID = -1; + + public static final long REGISTRATION_TIMEOUT = 500000; + + public static final byte OPCODE_REGISTRATION = 1; + + public static final byte OPCODE_REGISTRATION_ACK = 2; + + public static final byte OPCODE_DEREGISTRATION = 3; + + public static final byte SUCCESS = 1; + + public static final byte FAILURE = 0; + + private SynchronizingCorrelator<Short, Boolean> registrations = new SynchronizingCorrelator(); + + public ControlChannelImpl(AbstractTCPConnector connector) + { + super(connector.getReceiveExecutor()); + setChannelID(CONTROL_CHANNEL_ID); + setConnector(connector); + } + + public boolean registerChannel(short channelID, String protocolID) + { + assertValidChannelID(channelID); + Synchronizer<Boolean> registration = registrations.correlate(channelID); + + Buffer buffer = provideBuffer(); + ByteBuffer byteBuffer = buffer.startPutting(CONTROL_CHANNEL_ID); + byteBuffer.put(OPCODE_REGISTRATION); + byteBuffer.putShort(channelID); + BufferUtil.putUTF8(byteBuffer, protocolID); + handleBuffer(buffer); + + return registration.get(REGISTRATION_TIMEOUT); + } + + public void deregisterChannel(short channelID) + { + assertValidChannelID(channelID); + + Buffer buffer = provideBuffer(); + ByteBuffer byteBuffer = buffer.startPutting(CONTROL_CHANNEL_ID); + byteBuffer.put(OPCODE_DEREGISTRATION); + byteBuffer.putShort(channelID); + handleBuffer(buffer); + } + + private void assertValidChannelID(short channelID) + { + if (channelID <= CONTROL_CHANNEL_ID) + { + throw new IllegalArgumentException("channelID <= CONTROL_CHANNEL_ID"); + } + } + + public void handleBufferFromMultiplexer(Buffer buffer) + { + try + { + ByteBuffer byteBuffer = buffer.getByteBuffer(); + byte opcode = byteBuffer.get(); + System.out.println("CONTROL: " + opcode); + switch (opcode) + { + case OPCODE_REGISTRATION: + { + short channelID = byteBuffer.getShort(); + assertValidChannelID(channelID); + boolean success = true; + + try + { + byte[] handlerFactoryUTF8 = BufferUtil.getByteArray(byteBuffer); + String protocolID = BufferUtil.fromUTF8(handlerFactoryUTF8); + ChannelImpl channel = ((AbstractTCPConnector)getConnector()).createChannel(channelID, + protocolID); + if (channel != null) + { + channel.activate(); + } + else + { + success = false; + } + } + catch (Exception ex) + { + ex.printStackTrace(); + success = false; + } + + sendStatus(OPCODE_REGISTRATION_ACK, channelID, success); + break; + } + + case OPCODE_REGISTRATION_ACK: + { + short channelID = byteBuffer.getShort(); + boolean success = byteBuffer.get() == SUCCESS; + registrations.put(channelID, success); + break; + } + + case OPCODE_DEREGISTRATION: + throw new UnsupportedOperationException(); + + default: + System.out.println("Invalid opcode: " + opcode); + ((AbstractTCPConnector)getConnector()).deactivate(); + break; + } + } + finally + { + buffer.release(); + } + } + + // private Buffer getBuffer() + // { + // return + // ((AbstractTCPConnector)getConnector()).getBufferProvider().provideBuffer(); + // } + + private void sendStatus(byte opcode, short channelID, boolean status) + { + Buffer buffer = provideBuffer(); + ByteBuffer byteBuffer = buffer.startPutting(CONTROL_CHANNEL_ID); + byteBuffer.put(opcode); + byteBuffer.putShort(channelID); + byteBuffer.put(status ? SUCCESS : FAILURE); + handleBuffer(buffer); + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/tcp/ServerTCPConnectorImpl.java b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/tcp/ServerTCPConnectorImpl.java new file mode 100644 index 0000000000..f79e8aeb6f --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/tcp/ServerTCPConnectorImpl.java @@ -0,0 +1,72 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.internal.net4j.transport.tcp; + +import org.eclipse.net4j.transport.BufferProvider; +import org.eclipse.net4j.transport.ProtocolFactory; +import org.eclipse.net4j.transport.tcp.TCPSelector; +import org.eclipse.net4j.util.registry.IRegistry; + +import java.net.SocketAddress; +import java.nio.channels.SocketChannel; +import java.util.concurrent.ExecutorService; + +/** + * @author Eike Stepper + */ +public class ServerTCPConnectorImpl extends AbstractTCPConnector +{ + public ServerTCPConnectorImpl(SocketChannel socketChannel, ExecutorService receiveExecutor, + IRegistry<String, ProtocolFactory> protocolFactoryRegistry, BufferProvider bufferProvider, + TCPSelector selector) + { + super(socketChannel); + setReceiveExecutor(receiveExecutor); + setProtocolFactoryRegistry(protocolFactoryRegistry); + setBufferProvider(bufferProvider); + setSelector(selector); + } + + public Type getType() + { + return Type.SERVER; + } + + public String getHost() + { + return getSocketChannel().socket().getInetAddress().getHostAddress(); + } + + public int getPort() + { + return getSocketChannel().socket().getPort(); + } + + @Override + public String toString() + { + try + { + SocketAddress address = getSocketChannel().socket().getRemoteSocketAddress(); + return "ServerTCPConnector[" + address + "]"; + } + catch (Exception ex) + { + return "ServerTCPConnector"; + } + } + + @Override + protected void onDeactivate() throws Exception + { + super.onDeactivate(); + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/tcp/TCPAcceptorImpl.java b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/tcp/TCPAcceptorImpl.java new file mode 100644 index 0000000000..801c281c32 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/tcp/TCPAcceptorImpl.java @@ -0,0 +1,371 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.internal.net4j.transport.tcp; + +import org.eclipse.net4j.transport.Buffer; +import org.eclipse.net4j.transport.BufferProvider; +import org.eclipse.net4j.transport.ProtocolFactory; +import org.eclipse.net4j.transport.tcp.TCPAcceptor; +import org.eclipse.net4j.transport.tcp.TCPAcceptorListener; +import org.eclipse.net4j.transport.tcp.TCPConnector; +import org.eclipse.net4j.transport.tcp.TCPSelector; +import org.eclipse.net4j.transport.tcp.TCPSelectorListener; +import org.eclipse.net4j.util.lifecycle.AbstractLifecycle; +import org.eclipse.net4j.util.lifecycle.LifecycleListener; +import org.eclipse.net4j.util.lifecycle.LifecycleNotifier; +import org.eclipse.net4j.util.lifecycle.LifecycleUtil; +import org.eclipse.net4j.util.registry.IRegistry; + +import org.eclipse.internal.net4j.transport.ChannelImpl; + +import java.io.IOException; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.nio.channels.Channel; +import java.nio.channels.ClosedByInterruptException; +import java.nio.channels.SelectionKey; +import java.nio.channels.ServerSocketChannel; +import java.nio.channels.SocketChannel; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.Queue; +import java.util.Set; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.ExecutorService; + +/** + * @author Eike Stepper + */ +public class TCPAcceptorImpl extends AbstractLifecycle implements TCPAcceptor, BufferProvider, + TCPSelectorListener.Passive, LifecycleListener +{ + private IRegistry<String, ProtocolFactory> protocolFactoryRegistry; + + private BufferProvider bufferProvider; + + private TCPSelector selector; + + private String listenAddr = DEFAULT_ADDRESS; + + private int listenPort = DEFAULT_PORT; + + private ServerSocketChannel serverSocketChannel; + + private SelectionKey selectionKey; + + private Set<TCPConnector> acceptedConnectors = new HashSet(); + + /** + * An optional executor to be used by the {@link Channel}s to process their + * {@link ChannelImpl#receiveQueue} instead of the current thread. If not + * <code>null</code> the calling thread of + * {@link ChannelImpl#handleBufferFromMultiplexer(Buffer)} becomes decoupled. + * <p> + */ + private ExecutorService receiveExecutor; + + /** + * Don't initialize lazily to circumvent synchronization! + */ + private Queue<TCPAcceptorListener> listeners = new ConcurrentLinkedQueue(); + + public TCPAcceptorImpl() + { + } + + public ExecutorService getReceiveExecutor() + { + return receiveExecutor; + } + + public void setReceiveExecutor(ExecutorService receiveExecutor) + { + this.receiveExecutor = receiveExecutor; + } + + public IRegistry<String, ProtocolFactory> getProtocolFactoryRegistry() + { + return protocolFactoryRegistry; + } + + public void setProtocolFactoryRegistry(IRegistry<String, ProtocolFactory> protocolFactoryRegistry) + { + this.protocolFactoryRegistry = protocolFactoryRegistry; + } + + public BufferProvider getBufferProvider() + { + return bufferProvider; + } + + public void setBufferProvider(BufferProvider bufferProvider) + { + this.bufferProvider = bufferProvider; + } + + public short getBufferCapacity() + { + return bufferProvider.getBufferCapacity(); + } + + public Buffer provideBuffer() + { + return bufferProvider.provideBuffer(); + } + + public void retainBuffer(Buffer buffer) + { + bufferProvider.retainBuffer(buffer); + } + + public TCPSelector getSelector() + { + return selector; + } + + public void setSelector(TCPSelector selector) + { + this.selector = selector; + } + + public String getAddress() + { + return listenAddr; + } + + public void setListenAddr(String listenAddr) + { + this.listenAddr = listenAddr; + } + + public int getPort() + { + return listenPort; + } + + public void setListenPort(int listenPort) + { + this.listenPort = listenPort; + } + + public TCPConnector[] getAcceptedConnectors() + { + ArrayList<TCPConnector> result; + synchronized (acceptedConnectors) + { + result = new ArrayList<TCPConnector>(acceptedConnectors); + } + + return result.toArray(new TCPConnector[result.size()]); + } + + public void addListener(TCPAcceptorListener listener) + { + listeners.add(listener); + } + + public void removeListener(TCPAcceptorListener listener) + { + listeners.remove(listener); + } + + public void notifyLifecycleActivated(LifecycleNotifier notifier) + { + // Do nothing + } + + public void notifyLifecycleDeactivating(LifecycleNotifier notifier) + { + synchronized (acceptedConnectors) + { + notifier.removeLifecycleListener(this); + acceptedConnectors.remove(notifier); + } + } + + public void handleAccept(TCPSelector selector, ServerSocketChannel serverSocketChannel) + { + try + { + SocketChannel socketChannel = serverSocketChannel.accept(); + if (socketChannel != null) + { + socketChannel.configureBlocking(false); + addConnector(socketChannel); + } + } + catch (ClosedByInterruptException ex) + { + deactivate(); + } + catch (Exception ex) + { + if (isActive()) + { + ex.printStackTrace(); + } + + deactivate(); + } + } + + @Override + public String toString() + { + return "TCPAcceptor[" + "/" + listenAddr + ":" + listenPort + "]"; + } + + protected void addConnector(SocketChannel socketChannel) + { + try + { + AbstractTCPConnector connector = createConnector(socketChannel); + connector.activate(); + connector.addLifecycleListener(this); + + synchronized (acceptedConnectors) + { + acceptedConnectors.add(connector); + } + + fireConnectorAccepted(connector); + } + catch (Exception ex) + { + ex.printStackTrace(); + + try + { + socketChannel.close(); + } + catch (IOException ioex) + { + ioex.printStackTrace(); + } + } + } + + protected AbstractTCPConnector createConnector(SocketChannel socketChannel) + { + return new ServerTCPConnectorImpl(socketChannel, getReceiveExecutor(), + getProtocolFactoryRegistry(), bufferProvider, selector); + } + + protected void fireConnectorAccepted(TCPConnector connector) + { + for (TCPAcceptorListener listener : listeners) + { + try + { + listener.notifyConnectorAccepted(this, connector); + } + catch (Exception ex) + { + ex.printStackTrace(); + } + } + } + + @Override + protected void onAccessBeforeActivate() throws Exception + { + super.onAccessBeforeActivate(); + if (bufferProvider == null) + { + throw new IllegalStateException("bufferProvider == null"); + } + + if (protocolFactoryRegistry == null) + { + System.out.println(toString() + ": (INFO) protocolFactoryRegistry == null"); + } + + if (receiveExecutor == null) + { + System.out.println(toString() + ": (INFO) receiveExecutor == null"); + } + + if (selector == null) + { + selector = TCPUtil.createTCPSelector(); + LifecycleUtil.activate(selector); + } + } + + @Override + protected void onActivate() throws Exception + { + super.onActivate(); + InetAddress addr = InetAddress.getByName(listenAddr); + InetSocketAddress sAddr = new InetSocketAddress(addr, listenPort); + + serverSocketChannel = ServerSocketChannel.open(); + serverSocketChannel.configureBlocking(false); + serverSocketChannel.socket().bind(sAddr); + + selectionKey = selector.register(serverSocketChannel, this); + } + + @Override + protected void onDeactivate() throws Exception + { + for (TCPConnector connector : getAcceptedConnectors()) + { + try + { + LifecycleUtil.deactivate(connector); + } + catch (Exception ex) + { + ex.printStackTrace(); + } + } + + Exception exception = null; + + try + { + selectionKey.cancel(); + } + catch (RuntimeException ex) + { + if (exception == null) + { + exception = ex; + } + } + finally + { + selectionKey = null; + } + + try + { + serverSocketChannel.close(); + } + catch (RuntimeException ex) + { + if (exception == null) + { + exception = ex; + } + } + finally + { + serverSocketChannel = null; + } + + if (exception != null) + { + throw exception; + } + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/tcp/TCPSelectorImpl.java b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/tcp/TCPSelectorImpl.java new file mode 100644 index 0000000000..ec1d1c66ed --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/tcp/TCPSelectorImpl.java @@ -0,0 +1,212 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.internal.net4j.transport.tcp; + +import org.eclipse.net4j.transport.tcp.TCPSelector; +import org.eclipse.net4j.transport.tcp.TCPSelectorListener; +import org.eclipse.net4j.util.lifecycle.AbstractLifecycle; + +import java.io.IOException; +import java.nio.channels.CancelledKeyException; +import java.nio.channels.ClosedChannelException; +import java.nio.channels.ClosedSelectorException; +import java.nio.channels.SelectableChannel; +import java.nio.channels.SelectionKey; +import java.nio.channels.Selector; +import java.nio.channels.ServerSocketChannel; +import java.nio.channels.SocketChannel; +import java.util.Iterator; + +/** + * @author Eike Stepper + */ +public class TCPSelectorImpl extends AbstractLifecycle implements TCPSelector, Runnable +{ + private static final long SELECT_TIMEOUT = 100; + + private Selector selector; + + private Thread thread; + + public TCPSelectorImpl() + { + } + + public SelectionKey register(ServerSocketChannel channel, TCPSelectorListener.Passive listener) + throws ClosedChannelException + { + if (listener == null) + { + throw new IllegalArgumentException("listener == null"); + } + + System.out.println(toString() + ": Registering " + TCPUtil.toString(channel)); + return channel.register(selector, SelectionKey.OP_ACCEPT, listener); + } + + public SelectionKey register(SocketChannel channel, TCPSelectorListener.Active listener) + throws ClosedChannelException + { + if (listener == null) + { + throw new IllegalArgumentException("listener == null"); + } + + System.out.println(toString() + ": Registering " + TCPUtil.toString(channel)); + return channel.register(selector, SelectionKey.OP_CONNECT | SelectionKey.OP_READ, listener); + } + + public void run() + { + while (isActive()) + { + if (Thread.interrupted()) + { + deactivate(); + break; + } + + try + { + if (selector.select(SELECT_TIMEOUT) > 0) + { + Iterator<SelectionKey> it = selector.selectedKeys().iterator(); + while (it.hasNext()) + { + SelectionKey selKey = it.next(); + it.remove(); + + try + { + handleSelection(selKey); + } + catch (CancelledKeyException ignore) + { + ; // Do nothing + } + catch (Exception ex) + { + ex.printStackTrace(); + selKey.cancel(); + } + } + } + } + catch (ClosedSelectorException ex) + { + break; + } + catch (Exception ex) + { + ex.printStackTrace(); + deactivate(); + break; + } + } + } + + @Override + public String toString() + { + return "TCPSelector"; + } + + protected void handleSelection(SelectionKey selKey) throws IOException + { + SelectableChannel channel = selKey.channel(); + if (channel instanceof ServerSocketChannel) + { + ServerSocketChannel ssChannel = (ServerSocketChannel)selKey.channel(); + TCPSelectorListener.Passive listener = (TCPSelectorListener.Passive)selKey.attachment(); + + if (selKey.isAcceptable()) + { + System.out.println(toString() + ": Accepting " + TCPUtil.toString(ssChannel)); + listener.handleAccept(this, ssChannel); + } + } + else if (channel instanceof SocketChannel) + { + SocketChannel sChannel = (SocketChannel)channel; + TCPSelectorListener.Active listener = (TCPSelectorListener.Active)selKey.attachment(); + + if (selKey.isConnectable()) + { + System.out.println(toString() + ": Connecting " + TCPUtil.toString(sChannel)); + listener.handleConnect(this, sChannel); + } + + if (selKey.isReadable()) + { + System.out.println(toString() + ": Reading " + TCPUtil.toString(sChannel)); + listener.handleRead(this, sChannel); + } + + if (selKey.isWritable()) + { + System.out.println(toString() + ": Writing " + TCPUtil.toString(sChannel)); + listener.handleWrite(this, sChannel); + } + } + } + + @Override + protected void onActivate() throws Exception + { + selector = Selector.open(); + thread = new Thread(this); + thread.setDaemon(true); + thread.start(); + } + + @Override + protected void onDeactivate() throws Exception + { + Exception exception = null; + + try + { + thread.join(2 * SELECT_TIMEOUT); + } + catch (RuntimeException ex) + { + if (exception == null) + { + exception = ex; + } + } + finally + { + thread = null; + } + + try + { + selector.close(); + } + catch (Exception ex) + { + if (exception == null) + { + exception = ex; + } + } + finally + { + selector = null; + } + + if (exception != null) + { + throw exception; + } + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/tcp/TCPUtil.java b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/tcp/TCPUtil.java new file mode 100644 index 0000000000..634ca4389c --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/internal/net4j/transport/tcp/TCPUtil.java @@ -0,0 +1,161 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.internal.net4j.transport.tcp; + +import org.eclipse.net4j.transport.BufferProvider; +import org.eclipse.net4j.transport.tcp.TCPAcceptor; +import org.eclipse.net4j.transport.tcp.TCPConnector; +import org.eclipse.net4j.transport.tcp.TCPSelector; + +import java.nio.channels.SelectionKey; +import java.nio.channels.ServerSocketChannel; +import java.nio.channels.SocketChannel; + +/** + * @author Eike Stepper + */ +public final class TCPUtil +{ + private TCPUtil() + { + } + + public static TCPSelector createTCPSelector() + { + TCPSelectorImpl selector = new TCPSelectorImpl(); + return selector; + } + + public static TCPAcceptor createTCPAcceptor(BufferProvider bufferProvider, TCPSelector selector, + String address, int port) + { + TCPAcceptorImpl acceptor = new TCPAcceptorImpl(); + acceptor.setBufferProvider(bufferProvider); + acceptor.setSelector(selector); + acceptor.setListenPort(port); + acceptor.setListenAddr(address); + return acceptor; + } + + public static TCPAcceptor createTCPAcceptor(BufferProvider bufferProvider, TCPSelector selector) + { + return createTCPAcceptor(bufferProvider, selector, TCPAcceptor.DEFAULT_ADDRESS, + TCPAcceptor.DEFAULT_PORT); + } + + public static TCPConnector createTCPConnector(BufferProvider bufferProvider, + TCPSelector selector, String host, int port) + { + ClientTCPConnectorImpl connector = new ClientTCPConnectorImpl(); + connector.setBufferProvider(bufferProvider); + connector.setSelector(selector); + connector.setHost(host); + connector.setPort(port); + return connector; + } + + public static TCPConnector createTCPConnector(BufferProvider bufferProvider, + TCPSelector selector, String host) + { + return createTCPConnector(bufferProvider, selector, host, TCPConnector.DEFAULT_PORT); + } + + public static String toString(ServerSocketChannel channel) + { + return channel.toString(); + // return "ServerSocketChannel[" + channel.socket().getLocalSocketAddress() + // + "]"; + } + + public static String toString(SocketChannel channel) + { + return channel.toString(); + // return "SocketChannel[" + channel.socket().getRemoteSocketAddress() + + // "]"; + } + + public static String formatInterestOps(int newOps) + { + StringBuilder builder = new StringBuilder(); + if ((newOps & SelectionKey.OP_ACCEPT) != 0) + { + addInterestOp(builder, "ACCEPT"); + } + + if ((newOps & SelectionKey.OP_CONNECT) != 0) + { + addInterestOp(builder, "CONNECT"); + } + + if ((newOps & SelectionKey.OP_READ) != 0) + { + addInterestOp(builder, "READ"); + } + + if ((newOps & SelectionKey.OP_WRITE) != 0) + { + addInterestOp(builder, "WRITE"); + } + + return builder.toString(); + } + + public static void setInterest(SelectionKey selectionKey, int operation, boolean interested) + { + int newOps; + int oldOps = selectionKey.interestOps(); + if (interested) + { + newOps = oldOps | operation; + } + else + { + newOps = oldOps & ~operation; + } + + if (oldOps != newOps) + { + System.out.println(selectionKey.channel().toString() + ": Setting interest " + + formatInterestOps(newOps) + " (was " + formatInterestOps(oldOps).toLowerCase() + ")"); + selectionKey.interestOps(newOps); + } + } + + public static void setAcceptInterest(SelectionKey selectionKey, boolean interested) + { + setInterest(selectionKey, SelectionKey.OP_ACCEPT, interested); + } + + public static void setConnectInterest(SelectionKey selectionKey, boolean interested) + { + setInterest(selectionKey, SelectionKey.OP_CONNECT, interested); + } + + public static void setReadInterest(SelectionKey selectionKey, boolean interested) + { + setInterest(selectionKey, SelectionKey.OP_READ, interested); + } + + public static void setWriteInterest(SelectionKey selectionKey, boolean interested) + { + setInterest(selectionKey, SelectionKey.OP_WRITE, interested); + } + + private static void addInterestOp(StringBuilder builder, String op) + { + if (builder.length() != 0) + { + builder.append("|"); + } + + builder.append(op); + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/message/Deserializer.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/message/Deserializer.java new file mode 100644 index 0000000000..b1a50a5d9c --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/message/Deserializer.java @@ -0,0 +1,22 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.message; + +/** + * TODO The {@link Deserializer} class. + * <p> + * + * @author Eike Stepper + */ +public interface Deserializer +{ + +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/message/Serializer.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/message/Serializer.java new file mode 100644 index 0000000000..fb6e4f7594 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/message/Serializer.java @@ -0,0 +1,22 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.message; + +/** + * TODO The {@link Serializer} class. + * <p> + * + * @author Eike Stepper + */ +public interface Serializer +{ + +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/remote/ServiceExporter.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/remote/ServiceExporter.java new file mode 100644 index 0000000000..824951cee2 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/remote/ServiceExporter.java @@ -0,0 +1,22 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.remote; + +/** + * TODO The {@link ServiceExporter} class. + * <p> + * + * @author Eike Stepper + */ +public interface ServiceExporter +{ + +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/remote/ServiceImporter.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/remote/ServiceImporter.java new file mode 100644 index 0000000000..bf4b350c86 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/remote/ServiceImporter.java @@ -0,0 +1,22 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.remote; + +/** + * TODO The {@link ServiceImporter} class. + * <p> + * + * @author Eike Stepper + */ +public interface ServiceImporter +{ + +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/signal/Indication.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/signal/Indication.java new file mode 100644 index 0000000000..a91fefe844 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/signal/Indication.java @@ -0,0 +1,32 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.signal; + +import java.io.InputStream; + +/** + * @author Eike Stepper + */ +public abstract class Indication extends Signal +{ + protected Indication() + { + } + + @Override + public String toString() + { + return "Indication[" + getSignalID() + ", " + getProtocol() + ", correlation=" + + getCorrelationID() + "]"; + } + + protected abstract void indicating(InputStream stream) throws Exception; +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/signal/Request.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/signal/Request.java new file mode 100644 index 0000000000..f9f75314c2 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/signal/Request.java @@ -0,0 +1,39 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.signal; + +import java.io.OutputStream; + +/** + * @author Eike Stepper + */ +public abstract class Request extends Signal +{ + protected Request(SignalProtocol protocol) + { + setProtocol(protocol); + setCorrelationID(protocol.getNextCorrelationID()); + } + + public Object send() throws Exception + { + return getProtocol().sendRequest(this); + } + + @Override + public String toString() + { + return "Request[" + getSignalID() + ", " + getProtocol() + ", correlation=" + + getCorrelationID() + "]"; + } + + protected abstract void requesting(OutputStream stream) throws Exception; +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/signal/Signal.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/signal/Signal.java new file mode 100644 index 0000000000..9fffa816ed --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/signal/Signal.java @@ -0,0 +1,47 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.signal; + +/** + * @author Eike Stepper + */ +public abstract class Signal +{ + private SignalProtocol protocol; + + private int correlationID; + + protected Signal() + { + } + + protected SignalProtocol getProtocol() + { + return protocol; + } + + protected int getCorrelationID() + { + return correlationID; + } + + protected abstract short getSignalID(); + + void setProtocol(SignalProtocol protocol) + { + this.protocol = protocol; + } + + void setCorrelationID(int correlationID) + { + this.correlationID = correlationID; + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/signal/SignalProtocol.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/signal/SignalProtocol.java new file mode 100644 index 0000000000..b65bd9cd18 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/signal/SignalProtocol.java @@ -0,0 +1,70 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.signal; + +import org.eclipse.net4j.transport.Buffer; +import org.eclipse.net4j.transport.Channel; + +import org.eclipse.internal.net4j.transport.AbstractProtocol; + +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; + +/** + * @author Eike Stepper + */ +public abstract class SignalProtocol extends AbstractProtocol +{ + private int nextCorrelationID; + + private Queue<Request> requestQueue = new ConcurrentLinkedQueue(); + + public SignalProtocol(Channel channel) + { + super(channel); + } + + public void handleBuffer(Buffer buffer) + { + // TODO Implement method SignalProtocol + + } + + @Override + public String toString() + { + return "SignalProtocol[" + getProtocolID() + ", " + getChannel() + "]"; + } + + protected abstract Indication createIndication(short signalID); + + Object sendRequest(Request request) + { + // TODO Implement method SignalProtocol + return null; + } + + int getNextCorrelationID() + { + int correlationID = nextCorrelationID; + if (nextCorrelationID == Integer.MAX_VALUE) + { + System.out.println(toString() + ": Correlation wrap around"); + nextCorrelationID = 0; + } + else + { + ++nextCorrelationID; + } + + return correlationID; + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/Buffer.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/Buffer.java new file mode 100644 index 0000000000..1b095544bf --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/Buffer.java @@ -0,0 +1,104 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.transport; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.channels.SocketChannel; + +/** + * Basic <b>unit of transport</b> in Net4j. A buffer is well prepared for the + * usage with asynchronous {@link Channel}s but can also be used with pure + * {@link SocketChannel}s. All methods of <code>Buffer</code> are + * non-blocking. + * <p> + * Usually buffers are obtained from a {@link BufferProvider}. Buffers can be + * acessed, passed around and finally {@link #release() released} to their + * original provider. The capacity of a buffer is determined by its provider. + * <p> + * In addition to its payload data each buffer contains an internal header of + * four bytes, two of them representing a channel identifier the other two of + * them denoting the length of the payload data. The payload data may be + * accessed through a {@link #getByteBuffer() ByteBuffer} (TODO see restrictions + * below). + * <p> + * An example for <b>putting</b> values into a buffer and writing it to a + * {@link SocketChannel}: + * <p> + * + * <pre> + * // Obtain a fresh buffer + * Buffer buffer = bufferProvider.getBuffer(); + * + * // Start filling the buffer for channelID 4711 + * ByteBuffer byteBuffer = buffer.startPutting(4711); + * byteBuffer.putDouble(15.47); + * + * // Write the contents of the Buffer to a + * // SocketChannel without blocking + * while (!buffer.write(socketChannel)) + * { + * // Do something else + * } + * </pre> + * + * An example for reading a buffer from a {@link SocketChannel} and <b>getting</b> + * values from it: + * <p> + * + * <pre> + * // Obtain a fresh buffer + * Buffer buffer = bufferProvider.getBuffer(); + * + * // Read the contents of the Buffer from a + * // SocketChannel without blocking + * ByteBuffer byteBuffer; + * while ((byteBuffer = buffer.startGetting(socketChannel)) == null) + * { + * // Do something else + * } + * + * // Access the contents of the buffer and + * // release it to its provider + * double value = byteBuffer.getDouble(); + * buffer.release(); + * </pre> + * + * @see BufferProvider + * @see Channel#sendBuffer(Buffer) + * @see Channel#setReceiveHandler(BufferHandler) + * @see BufferHandler#handleBuffer(Buffer) + * @author Eike Stepper + */ +public interface Buffer +{ + public BufferProvider getBufferProvider(); + + public short getChannelID(); + + public short getCapacity(); + + public ByteBuffer startGetting(SocketChannel socketChannel) throws IOException; + + public ByteBuffer startPutting(short channelID); + + public boolean write(SocketChannel socketChannel) throws IOException; + + public ByteBuffer getByteBuffer(); + + public boolean isEOS(); + + public void setEOS(boolean eos); + + public void release(); + + public void clear(); +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/BufferHandler.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/BufferHandler.java new file mode 100644 index 0000000000..6b31450a9b --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/BufferHandler.java @@ -0,0 +1,16 @@ +package org.eclipse.net4j.transport; + +/** + * @author Eike Stepper + */ +public interface BufferHandler +{ + /** + * Handles a {@link Buffer} and optionally releases it. The implementor of + * this method takes over the ownership of the buffer. Care must be taken to + * properly {@link Buffer#release() release} the buffer if the ownership is + * not explicitely passed to some further party. + * <p> + */ + public void handleBuffer(Buffer buffer); +}
\ No newline at end of file diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/BufferPool.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/BufferPool.java new file mode 100644 index 0000000000..be9290b33b --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/BufferPool.java @@ -0,0 +1,19 @@ +package org.eclipse.net4j.transport; + +/** + * @author Eike Stepper + */ +public interface BufferPool extends BufferProvider +{ + public boolean evictOne(); + + public int evict(int survivors); + + /** + * @author Eike Stepper + */ + public interface Introspection extends BufferProvider.Introspection + { + public int getPooledBuffers(); + } +}
\ No newline at end of file diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/BufferProvider.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/BufferProvider.java new file mode 100644 index 0000000000..47b42ecb0b --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/BufferProvider.java @@ -0,0 +1,23 @@ +package org.eclipse.net4j.transport; + +/** + * @author Eike Stepper + */ +public interface BufferProvider +{ + public short getBufferCapacity(); + + public Buffer provideBuffer(); + + public void retainBuffer(Buffer buffer); + + /** + * @author Eike Stepper + */ + public interface Introspection + { + public long getProvidedBuffers(); + + public long getRetainedBuffers(); + } +}
\ No newline at end of file diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/Channel.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/Channel.java new file mode 100644 index 0000000000..c57af778c1 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/Channel.java @@ -0,0 +1,38 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.transport; + +/** + * A bidirectional communications channel for the asynchronous exchange of + * {@link Buffer}s. A channel is lightweight and virtual in the sense that it + * does not necessarily represent a single physical connection like a TCP socket + * connection. The underlying physical connection is represented by a + * {@link Connector}. + * <p> + * This interface is <b>not</b> intended to be implemented by clients. + * <p> + * + * @author Eike Stepper + */ +public interface Channel extends BufferHandler +{ + public short getChannelID(); + + public Connector getConnector(); + + public void sendBuffer(Buffer buffer); + + public BufferHandler getReceiveHandler(); + + public void setReceiveHandler(BufferHandler receiveHandler); + + public void close(); +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/Connector.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/Connector.java new file mode 100644 index 0000000000..617047eead --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/Connector.java @@ -0,0 +1,157 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.transport; + +import org.eclipse.net4j.util.registry.IRegistry; + +/** + * One endpoint of a physical connection of arbitrary nature between two + * communicating parties. A {@link Connector} encapsulates the process of + * establishing and closing such connections and has a {@link Type} of + * {@link Type#CLIENT FOR_CLIENTS} or {@link Type#SERVER FOR_SERVERS} with + * respect to this process. Once a connection is established either party can + * use its {@link Connector} to open multiple {@link Channel}s to + * asynchronously exchange {@link Buffer}s. + * <p> + * This interface is <b>not</b> intended to be implemented by clients. + * <p> + * + * @author Eike Stepper + */ +public interface Connector +{ + public Type getType(); + + public boolean isClient(); + + public boolean isServer(); + + public ConnectorCredentials getCredentials(); + + public void setCredentials(ConnectorCredentials credentials); + + public State getState(); + + /** + * Same as <code>{@link #getState()} == {@link State#CONNECTED}</code>. + * <p> + */ + public boolean isConnected(); + + /** + * Asynchronous connect. May leave this {@link Connector} in a state where + * <code>{@link #isConnected()} == false</code>. + * <p> + */ + public void connectAsync() throws ConnectorException; + + /** + * Blocks until <code>{@link #isConnected()} == true</code>. + * <p> + * + * @throws ConnectorException + */ + public boolean waitForConnection(long timeout) throws ConnectorException; + + /** + * Synchronous connect. Blocks until <code>{@link #isConnected()} == + * true</code>. + * <p> + */ + public boolean connect(long timeout) throws ConnectorException; + + public ConnectorException disconnect(); + + public Channel[] getChannels(); + + /** + * Synchronous request to open a new {@link Channel} with an undefined channel + * protocol. Since the peer connector can't lookup a {@link ProtocolFactory} + * without a protocol identifier the {@link BufferHandler} of the peer + * {@link Channel} can only be provided by external {@link ChannelListener}s. + * <p> + * + * @see #openChannel(String) + */ + public Channel openChannel() throws ConnectorException; + + /** + * Synchronous request to open a new {@link Channel} with a channel protocol + * defined by a given protocol identifier. The peer connector will lookup a + * {@link ProtocolFactory} with the protocol identifier, create a + * {@link BufferHandler} and inject it into the peer {@link Channel}. + * <p> + * + * @see #openChannel() + */ + public Channel openChannel(String protocolID) throws ConnectorException; + + /** + * Returns the {@link IRegistry} of {@link ProtocolFactory}s associated with + * this connector. + * <p> + * + * @see #setProtocolFactoryRegistry(IRegistry) + * @return The registry or <code>null</code>. + */ + public IRegistry<String, ProtocolFactory> getProtocolFactoryRegistry(); + + /** + * Sets the {@link IRegistry} of {@link ProtocolFactory}s for this connector. + * <p> + * + * @param protocolFactoryRegistry + * The registry or <code>null</code>. + */ + public void setProtocolFactoryRegistry(IRegistry<String, ProtocolFactory> protocolFactoryRegistry); + + public void addStateListener(StateListener listener); + + public void removeStateListener(StateListener listener); + + public void addChannelListener(ChannelListener listener); + + public void removeChannelListener(ChannelListener listener); + + /** + * @author Eike Stepper + */ + public enum Type + { + CLIENT, SERVER + } + + /** + * @author Eike Stepper + */ + public enum State + { + DISCONNECTED, CONNECTING, NEGOTIATING, CONNECTED + } + + /** + * @author Eike Stepper + */ + public interface StateListener + { + public void notifyStateChanged(Connector connector, State newState, State oldState); + } + + /** + * @author Eike Stepper + */ + public interface ChannelListener + { + public void notifyChannelOpened(Channel channel); + + public void notifyChannelClosing(Channel channel); + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/ConnectorCredentials.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/ConnectorCredentials.java new file mode 100644 index 0000000000..0a70b1cc0b --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/ConnectorCredentials.java @@ -0,0 +1,19 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.transport; + +/** + * @author Eike Stepper + */ +public interface ConnectorCredentials +{ + public String getUserID(); +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/ConnectorException.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/ConnectorException.java new file mode 100644 index 0000000000..6c9faa4f22 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/ConnectorException.java @@ -0,0 +1,38 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.transport; + +/** + * @author Eike Stepper + */ +public class ConnectorException extends Exception +{ + private static final long serialVersionUID = 1L; + + public ConnectorException() + { + } + + public ConnectorException(String message) + { + super(message); + } + + public ConnectorException(Throwable cause) + { + super(cause); + } + + public ConnectorException(String message, Throwable cause) + { + super(message, cause); + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/ConnectorNegotiator.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/ConnectorNegotiator.java new file mode 100644 index 0000000000..14f4c60ffa --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/ConnectorNegotiator.java @@ -0,0 +1,28 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.transport; + +/** + * @author Eike Stepper + */ +public interface ConnectorNegotiator +{ + /** + * Executes a negotitation phase between two connectors. + * <p> + * + * @param connector + * The connector to be used in the negotiation phase. + * @return <code>true</code> if the negotiation succeeded, + * <code>false</code> otherwise. + */ + public boolean negotiate(Connector connector); +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/Protocol.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/Protocol.java new file mode 100644 index 0000000000..3a7a1c83ec --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/Protocol.java @@ -0,0 +1,23 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.transport; + +/** + * @author Eike Stepper + */ +public interface Protocol extends BufferHandler +{ + public String getProtocolID(); + + public Channel getChannel(); + + public void dispose(); +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/ProtocolFactory.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/ProtocolFactory.java new file mode 100644 index 0000000000..86f6c3877a --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/ProtocolFactory.java @@ -0,0 +1,32 @@ +package org.eclipse.net4j.transport; + +import org.eclipse.net4j.transport.Connector.Type; +import org.eclipse.net4j.util.registry.IRegistryElement; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashSet; +import java.util.Set; + +/** + * @author Eike Stepper + */ +public interface ProtocolFactory extends IRegistryElement<String> +{ + public static final Set<Type> FOR_CLIENTS = Collections.singleton(Type.CLIENT); + + public static final Set<Type> FOR_SERVERS = Collections.singleton(Type.SERVER); + + public static final Set<Type> SYMMETRIC = Collections.unmodifiableSet(new HashSet<Type>(Arrays + .asList(new Type[] { Type.CLIENT, Type.SERVER }))); + + public Set<Type> getConnectorTypes(); + + public boolean isForClients(); + + public boolean isForServers(); + + public boolean isSymmetric(); + + public Protocol createProtocol(Channel channel); +}
\ No newline at end of file diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/tcp/TCPAcceptor.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/tcp/TCPAcceptor.java new file mode 100644 index 0000000000..bf4117d878 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/tcp/TCPAcceptor.java @@ -0,0 +1,44 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.transport.tcp; + +import org.eclipse.net4j.transport.ProtocolFactory; +import org.eclipse.net4j.util.registry.IRegistry; + +import java.util.concurrent.ExecutorService; + +/** + * @author Eike Stepper + */ +public interface TCPAcceptor +{ + public static final String DEFAULT_ADDRESS = "0.0.0.0"; + + public static final int DEFAULT_PORT = 2036; + + public String getAddress(); + + public int getPort(); + + public TCPConnector[] getAcceptedConnectors(); + + public IRegistry<String, ProtocolFactory> getProtocolFactoryRegistry(); + + public void setProtocolFactoryRegistry(IRegistry<String, ProtocolFactory> protocolFactoryRegistry); + + public void addListener(TCPAcceptorListener listener); + + public void removeListener(TCPAcceptorListener listener); + + public ExecutorService getReceiveExecutor(); + + public void setReceiveExecutor(ExecutorService receiveExecutor); +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/tcp/TCPAcceptorListener.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/tcp/TCPAcceptorListener.java new file mode 100644 index 0000000000..00b067a13a --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/tcp/TCPAcceptorListener.java @@ -0,0 +1,9 @@ +package org.eclipse.net4j.transport.tcp; + +/** + * @author Eike Stepper + */ +public interface TCPAcceptorListener +{ + public void notifyConnectorAccepted(TCPAcceptor acceptor, TCPConnector connector); +}
\ No newline at end of file diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/tcp/TCPConnector.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/tcp/TCPConnector.java new file mode 100644 index 0000000000..536592296f --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/tcp/TCPConnector.java @@ -0,0 +1,25 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.transport.tcp; + +import org.eclipse.net4j.transport.Connector; + +/** + * @author Eike Stepper + */ +public interface TCPConnector extends Connector +{ + public static final int DEFAULT_PORT = TCPAcceptor.DEFAULT_PORT; + + public int getPort(); + + public String getHost(); +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/tcp/TCPSelector.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/tcp/TCPSelector.java new file mode 100644 index 0000000000..80a4df9fb1 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/tcp/TCPSelector.java @@ -0,0 +1,28 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.transport.tcp; + +import java.nio.channels.ClosedChannelException; +import java.nio.channels.SelectionKey; +import java.nio.channels.ServerSocketChannel; +import java.nio.channels.SocketChannel; + +/** + * @author Eike Stepper + */ +public interface TCPSelector +{ + public SelectionKey register(ServerSocketChannel channel, TCPSelectorListener.Passive listener) + throws ClosedChannelException; + + public SelectionKey register(SocketChannel channel, TCPSelectorListener.Active listener) + throws ClosedChannelException; +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/tcp/TCPSelectorListener.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/tcp/TCPSelectorListener.java new file mode 100644 index 0000000000..1e95c6b023 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/transport/tcp/TCPSelectorListener.java @@ -0,0 +1,30 @@ +package org.eclipse.net4j.transport.tcp; + +import java.nio.channels.ServerSocketChannel; +import java.nio.channels.SocketChannel; + +/** + * @author Eike Stepper + */ +public interface TCPSelectorListener +{ + /** + * @author Eike Stepper + */ + public interface Passive + { + public void handleAccept(TCPSelector selector, ServerSocketChannel serverSocketChannel); + } + + /** + * @author Eike Stepper + */ + public interface Active + { + public void handleConnect(TCPSelector selector, SocketChannel channel); + + public void handleRead(TCPSelector selector, SocketChannel socketChannel); + + public void handleWrite(TCPSelector selector, SocketChannel socketChannel); + } +}
\ No newline at end of file diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/concurrent/Asynchronizer.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/concurrent/Asynchronizer.java new file mode 100644 index 0000000000..60840f0777 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/concurrent/Asynchronizer.java @@ -0,0 +1,25 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.util.concurrent; + +/** + * @author Eike Stepper + */ +public class Asynchronizer +{ + public Asynchronizer() + { + } + + public void addWork(Runnable work) + { + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/concurrent/Correlator.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/concurrent/Correlator.java new file mode 100644 index 0000000000..959fbc3042 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/concurrent/Correlator.java @@ -0,0 +1,25 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.util.concurrent; + +/** + * @author Eike Stepper + */ +public interface Correlator<CORRELATION, VALUE> +{ + public boolean isCorrelated(CORRELATION correlation); + + public VALUE correlate(CORRELATION correlation); + + public VALUE correlateUnique(CORRELATION correlation); + + public VALUE uncorrelate(CORRELATION correlation); +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/concurrent/ResultSynchronizer.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/concurrent/ResultSynchronizer.java new file mode 100644 index 0000000000..a9cf6a9dec --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/concurrent/ResultSynchronizer.java @@ -0,0 +1,96 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.util.concurrent; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +/** + * @author Eike Stepper + */ +public final class ResultSynchronizer<RESULT> implements Synchronizer<RESULT> +{ + private RESULT result; + + private Object consumerLock = new Object(); + + private CountDownLatch producerLatch = new CountDownLatch(1); + + public ResultSynchronizer() + { + } + + public RESULT get(long timeout) + { + try + { + final long stop = System.currentTimeMillis() + timeout; + synchronized (consumerLock) + { + while (result == null) + { + try + { + final long remaining = stop - System.currentTimeMillis(); + if (remaining <= 0) + { + return null; + } + + consumerLock.wait(Math.min(remaining, 100)); + } + catch (InterruptedException ex) + { + return null; + } + } + + return result; + } + } + finally + { + producerLatch.countDown(); + } + } + + public void put(RESULT result) + { + synchronized (consumerLock) + { + this.result = result; + consumerLock.notifyAll(); + } + } + + public boolean put(RESULT result, long timeout) + { + synchronized (consumerLock) + { + this.result = result; + consumerLock.notifyAll(); + } + + try + { + if (!producerLatch.await(timeout, TimeUnit.MILLISECONDS)) + { + return false; + } + } + catch (InterruptedException ex) + { + return false; + } + + return true; + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/concurrent/Synchronizer.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/concurrent/Synchronizer.java new file mode 100644 index 0000000000..b8a432aa50 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/concurrent/Synchronizer.java @@ -0,0 +1,30 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.util.concurrent; + +/** + * Synchronizes a producer and a consumer thread letting the producer pass a + * value to the consumer. Both producer and consumer must have access to this + * {@link Synchronizer} and there must only ever exist one consumer for it. Once + * the result value is consumed by the consumer this {@link Synchronizer} must + * not be reused. + * <p> + * + * @author Eike Stepper + */ +public interface Synchronizer<RESULT> +{ + public RESULT get(long timeout); + + public void put(RESULT result); + + public boolean put(RESULT result, long timeout); +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/concurrent/SynchronizingCorrelator.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/concurrent/SynchronizingCorrelator.java new file mode 100644 index 0000000000..58f132fbda --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/concurrent/SynchronizingCorrelator.java @@ -0,0 +1,96 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.util.concurrent; + +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; + +/** + * @author Eike Stepper + */ +public class SynchronizingCorrelator<CORRELATION, RESULT> implements + Correlator<CORRELATION, Synchronizer<RESULT>> +{ + private ConcurrentMap<CORRELATION, Synchronizer<RESULT>> map = new ConcurrentHashMap(); + + public boolean isCorrelated(CORRELATION correlation) + { + return map.containsKey(correlation); + } + + public Synchronizer<RESULT> correlate(CORRELATION correlation) + { + Synchronizer<RESULT> synchronizer = map.get(correlation); + if (synchronizer == null) + { + synchronizer = createSynchronizer(correlation); + map.put(correlation, synchronizer); + } + + return synchronizer; + } + + public Synchronizer<RESULT> correlateUnique(CORRELATION correlation) + { + Synchronizer<RESULT> synchronizer = createSynchronizer(correlation); + if (map.putIfAbsent(correlation, synchronizer) != null) + { + throw new IllegalStateException("Already correlated: " + correlation); + } + + return synchronizer; + } + + public Synchronizer<RESULT> uncorrelate(CORRELATION correlation) + { + return map.remove(correlation); + } + + public RESULT get(CORRELATION correlation, long timeout) + { + return correlate(correlation).get(timeout); + } + + public void put(CORRELATION correlation, RESULT result) + { + correlate(correlation).put(result); + } + + public boolean put(CORRELATION correlation, RESULT result, long timeout) + { + return correlate(correlation).put(result, timeout); + } + + protected Synchronizer<RESULT> createSynchronizer(final CORRELATION correlation) + { + return new Synchronizer<RESULT>() + { + private Synchronizer<RESULT> delegate = new ResultSynchronizer<RESULT>(); + + public RESULT get(long timeout) + { + RESULT result = delegate.get(timeout); + uncorrelate(correlation); + return result; + } + + public void put(RESULT result) + { + delegate.put(result); + } + + public boolean put(RESULT result, long timeout) + { + return delegate.put(result, timeout); + } + }; + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/concurrent/Worker.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/concurrent/Worker.java new file mode 100644 index 0000000000..0a1d268b18 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/concurrent/Worker.java @@ -0,0 +1,25 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.util.concurrent; + +/** + * @author Eike Stepper + */ +public class Worker +{ + public Worker() + { + } + + public void addWork(Runnable work) + { + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/lifecycle/AbstractLifecycle.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/lifecycle/AbstractLifecycle.java new file mode 100644 index 0000000000..1fa2d4eb0d --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/lifecycle/AbstractLifecycle.java @@ -0,0 +1,115 @@ +package org.eclipse.net4j.util.lifecycle; + +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; + +/** + * @author Eike Stepper + */ +public abstract class AbstractLifecycle implements Lifecycle, LifecycleNotifier +{ + private boolean active; + + /** + * Don't initialize lazily to circumvent synchronization! + */ + private Queue<LifecycleListener> listeners = new ConcurrentLinkedQueue(); + + protected AbstractLifecycle() + { + } + + public final void addLifecycleListener(LifecycleListener listener) + { + listeners.add(listener); + } + + public final void removeLifecycleListener(LifecycleListener listener) + { + listeners.remove(listener); + } + + public final synchronized void activate() throws Exception + { + if (!active) + { + System.out.println(toString() + ": Activating"); + onAccessBeforeActivate(); + onActivate(); + active = true; + fireLifecycleActivated(); + } + } + + public final synchronized Exception deactivate() + { + if (active) + { + System.out.println(toString() + ": Deactivating"); + fireLifecycleDeactivating(); + + try + { + onDeactivate(); + } + catch (Exception ex) + { + ex.printStackTrace(); + return ex; + } + finally + { + active = false; + } + } + + return null; + } + + public final boolean isActive() + { + return active; + } + + protected void fireLifecycleActivated() + { + for (LifecycleListener listener : listeners) + { + try + { + listener.notifyLifecycleActivated(this); + } + catch (Exception ex) + { + ex.printStackTrace(); + } + } + } + + protected void fireLifecycleDeactivating() + { + for (LifecycleListener listener : listeners) + { + try + { + listener.notifyLifecycleDeactivating(this); + } + catch (Exception ex) + { + ex.printStackTrace(); + } + } + } + + protected void onAccessBeforeActivate() throws Exception + { + } + + protected void onActivate() throws Exception + { + } + + protected void onDeactivate() throws Exception + { + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/lifecycle/Activator.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/lifecycle/Activator.java new file mode 100644 index 0000000000..9294ecb754 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/lifecycle/Activator.java @@ -0,0 +1,13 @@ +package org.eclipse.net4j.util.lifecycle; + +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +@Retention(RetentionPolicy.RUNTIME) +@Target(ElementType.METHOD) +public @interface Activator +{ + boolean propagate() default true; +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/lifecycle/Deactivator.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/lifecycle/Deactivator.java new file mode 100644 index 0000000000..0781c7d0e7 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/lifecycle/Deactivator.java @@ -0,0 +1,13 @@ +package org.eclipse.net4j.util.lifecycle; + +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +@Retention(RetentionPolicy.RUNTIME) +@Target(ElementType.METHOD) +public @interface Deactivator +{ + boolean propagate() default true; +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/lifecycle/Lifecycle.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/lifecycle/Lifecycle.java new file mode 100644 index 0000000000..1556f4c222 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/lifecycle/Lifecycle.java @@ -0,0 +1,29 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.util.lifecycle; + +/** + * @author Eike Stepper + */ +public interface Lifecycle +{ + public void activate() throws Exception; + + public Exception deactivate(); + + /** + * @author Eike Stepper + */ + public interface Introspection + { + public boolean isActive(); + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/lifecycle/LifecycleListener.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/lifecycle/LifecycleListener.java new file mode 100644 index 0000000000..eaa3863ff6 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/lifecycle/LifecycleListener.java @@ -0,0 +1,11 @@ +package org.eclipse.net4j.util.lifecycle; + +/** + * @author Eike Stepper + */ +public interface LifecycleListener +{ + public void notifyLifecycleActivated(LifecycleNotifier notifier); + + public void notifyLifecycleDeactivating(LifecycleNotifier notifier); +}
\ No newline at end of file diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/lifecycle/LifecycleNotifier.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/lifecycle/LifecycleNotifier.java new file mode 100644 index 0000000000..f7f72be4d6 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/lifecycle/LifecycleNotifier.java @@ -0,0 +1,11 @@ +package org.eclipse.net4j.util.lifecycle; + +/** + * @author Eike Stepper + */ +public interface LifecycleNotifier extends Lifecycle.Introspection +{ + public void addLifecycleListener(LifecycleListener listener); + + public void removeLifecycleListener(LifecycleListener listener); +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/lifecycle/LifecycleUtil.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/lifecycle/LifecycleUtil.java new file mode 100644 index 0000000000..0eeec29f9a --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/lifecycle/LifecycleUtil.java @@ -0,0 +1,165 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.util.lifecycle; + +import java.lang.annotation.Annotation; +import java.lang.reflect.Method; + +/** + * @author Eike Stepper + */ +public final class LifecycleUtil +{ + private LifecycleUtil() + { + } + + public static boolean isActive(Object object) + { + if (object instanceof Lifecycle.Introspection) + { + return ((Lifecycle.Introspection)object).isActive(); + } + + return true; + } + + public static void activate(Object object) throws Exception + { + activate(object, false); + } + + /** + * @see Activator + */ + public static void activate(Object object, boolean useAnnotation) throws Exception + { + if (object instanceof Lifecycle) + { + ((Lifecycle)object).activate(); + } + else if (useAnnotation) + { + invokeAnnotation(object, Activator.class); + } + } + + public static Exception activateSilent(Object object) + { + return activateSilent(object, false); + } + + /** + * @see Activator + */ + public static Exception activateSilent(Object object, boolean useAnnotation) + { + try + { + activate(object, useAnnotation); + return null; + } + catch (Exception ex) + { + return ex; + } + } + + public static Exception deactivate(Object object) + { + return deactivate(object, false); + } + + /** + * @see Deactivator + */ + public static Exception deactivate(Object object, boolean useAnnotation) + { + if (object instanceof Lifecycle) + { + return ((Lifecycle)object).deactivate(); + } + else if (useAnnotation) + { + invokeAnnotation(object, Deactivator.class); + } + + return null; + } + + public static void deactivateNoisy(Object object) throws Exception + { + deactivateNoisy(object, false); + } + + public static void deactivateNoisy(Object object, boolean useAnnotation) throws Exception + { + Exception ex = deactivate(object, useAnnotation); + if (ex != null) + { + throw ex; + } + } + + private static void invokeAnnotation(Object object, Class annotationClass) + { + Class c = object.getClass(); + while (c != Object.class) + { + final Method[] methods = c.getDeclaredMethods(); + for (Method method : methods) + { + if (method.getParameterTypes().length == 0) + { + Annotation annotation = method.getAnnotation(annotationClass); + if (annotation != null) + { + invokeMethod(object, method); + boolean propagate = annotationClass == Activator.class ? ((Activator)annotation) + .propagate() : ((Deactivator)annotation).propagate(); + if (!propagate) + { + break; + } + } + } + } + + c = c.getSuperclass(); + } + } + + private static Object invokeMethod(Object object, Method method) + { + try + { + return method.invoke(object, (Object[])null); + } + catch (IllegalAccessException iae) + { + try + { + method.setAccessible(true); + return method.invoke(object, (Object[])null); + } + catch (Exception ex) + { + ex.printStackTrace(); + } + } + catch (Exception ex) + { + ex.printStackTrace(); + } + + return null; + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/map/AbstractCachingMap.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/map/AbstractCachingMap.java new file mode 100644 index 0000000000..530298e41c --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/map/AbstractCachingMap.java @@ -0,0 +1,323 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.util.map; + +import java.util.AbstractMap; +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** + * Implementation note: {@link AbstractCachingMap} does not preserve the + * "modifyable view" contract of {@link Map#entrySet()} as well as of + * {@link Map#keySet()}, i.e. they are disconnected sets and modifications + * applied to them are not applied to their underlying + * {@link AbstractCachingMap}. + * <p> + * + * @author Eike Stepper + */ +public abstract class AbstractCachingMap<K, V> extends AbstractDelegatingMap<K, V> +{ + protected AbstractCachingMap() + { + } + + protected AbstractCachingMap(Map<? extends K, ? extends V> t) + { + super(t); + } + + public void clear() + { + cachedClear(); + } + + public boolean containsKey(Object key) + { + return cachedContainsKey(key) || delegatedContainsKey(key); + } + + public boolean containsValue(Object value) + { + return cachedContainsValue(value) || cachedContainsValue(value); + } + + public Set<Entry<K, V>> entrySet() + { + return mergedEntrySet(); + } + + public V get(Object key) + { + V result = cachedGet(key); + if (result == null) + { + result = delegatedGet(key); + } + + return result; + } + + public boolean isEmpty() + { + return cachedIsEmpty() && delegatedIsEmpty(); + } + + public Set<K> keySet() + { + return mergedKeySet(); + } + + public V put(K key, V value) + { + return cachedPut(key, value); + } + + public void putAll(Map<? extends K, ? extends V> t) + { + cachedPutAll(t); + } + + public V remove(Object key) + { + return cachedRemove(key); + } + + public int size() + { + return keySet().size(); + } + + public Collection<V> values() + { + return mergedValues(); + } + + @Override + public boolean equals(Object obj) + { + return mergedEquals(obj); + } + + @Override + public int hashCode() + { + return mergedHashCode(); + } + + @Override + public String toString() + { + return mergedToString(); + } + + protected void cachedClear() + { + getCache().clear(); + } + + protected boolean cachedContainsKey(Object key) + { + return getCache().containsKey(key); + } + + protected boolean cachedContainsValue(Object value) + { + return getCache().containsValue(value); + } + + protected Set<Entry<K, V>> cachedEntrySet() + { + return getCache().entrySet(); + } + + protected V cachedGet(Object key) + { + return getCache().get(key); + } + + protected boolean cachedIsEmpty() + { + return getCache().isEmpty(); + } + + protected Set<K> cachedKeySet() + { + return getCache().keySet(); + } + + protected V cachedPut(K key, V value) + { + return getCache().put(key, value); + } + + protected void cachedPutAll(Map<? extends K, ? extends V> t) + { + getCache().putAll(t); + } + + protected V cachedRemove(Object key) + { + return getCache().remove(key); + } + + protected int cachedSize() + { + return getCache().size(); + } + + protected Collection<V> cachedValues() + { + return getCache().values(); + } + + protected boolean cachedEquals(Object obj) + { + return getCache().equals(obj); + } + + protected int cachedHashCode() + { + return getCache().hashCode(); + } + + protected String cachedToString() + { + return getCache().toString(); + } + + protected Set<Entry<K, V>> mergedEntrySet() + { + final Map<K, V> merged = new HashMap<K, V>(); + merged.putAll(getDelegate()); + merged.putAll(getCache()); + return merged.entrySet(); + } + + protected Set<K> mergedKeySet() + { + final Set<K> merged = new HashSet<K>(); + merged.addAll(getDelegate().keySet()); + merged.addAll(getCache().keySet()); + return merged; + } + + protected Collection<V> mergedValues() + { + final List<V> result = new ArrayList<V>(); + for (K key : keySet()) + { + result.add(get(key)); + } + + return result; + } + + /** + * @see AbstractMap#equals(Object) + */ + protected boolean mergedEquals(Object o) + { + if (o == this) + return true; + + if (!(o instanceof Map)) + return false; + + Map<K, V> t = (Map<K, V>)o; + if (t.size() != size()) + return false; + + try + { + Iterator<Entry<K, V>> i = entrySet().iterator(); + while (i.hasNext()) + { + Entry<K, V> e = i.next(); + K key = e.getKey(); + V value = e.getValue(); + if (value == null) + { + if (!(t.get(key) == null && t.containsKey(key))) + return false; + } + else + { + if (!value.equals(t.get(key))) + return false; + } + } + } + catch (ClassCastException unused) + { + return false; + } + catch (NullPointerException unused) + { + return false; + } + + return true; + } + + /** + * @see AbstractMap#hashCode() + */ + protected int mergedHashCode() + { + int h = 0; + Iterator<Entry<K, V>> i = entrySet().iterator(); + while (i.hasNext()) + h += i.next().hashCode(); + return h; + } + + /** + * @see AbstractMap#toString() + */ + protected String mergedToString() + { + StringBuffer buf = new StringBuffer(); + buf.append("{"); + + Iterator<Entry<K, V>> i = entrySet().iterator(); + boolean hasNext = i.hasNext(); + while (hasNext) + { + Entry<K, V> e = i.next(); + K key = e.getKey(); + V value = e.getValue(); + if (key == this) + buf.append("(this Map)"); + else + buf.append(key); + buf.append("="); + if (value == this) + buf.append("(this Map)"); + else + buf.append(value); + hasNext = i.hasNext(); + if (hasNext) + buf.append(", "); + } + + buf.append("}"); + return buf.toString(); + } + + protected abstract Map<K, V> getCache(); +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/map/AbstractDelegatingMap.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/map/AbstractDelegatingMap.java new file mode 100644 index 0000000000..ab445ceb70 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/map/AbstractDelegatingMap.java @@ -0,0 +1,192 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.util.map; + +import java.util.Collection; +import java.util.Map; +import java.util.Set; + +/** + * Implementation note: {@link AbstractDelegatingMap} does not necessarily + * preserve the "modifyable view" contract of {@link Map#entrySet()} as well as + * of {@link Map#keySet()}, i.e. they might be disconnected sets and + * modifications applied to them might not be applied to their underlying + * {@link AbstractDelegatingMap}. + * <p> + * + * @author Eike Stepper + */ +public abstract class AbstractDelegatingMap<K, V> implements Map<K, V> +{ + protected AbstractDelegatingMap() + { + } + + protected AbstractDelegatingMap(Map<? extends K, ? extends V> t) + { + putAll(t); + } + + public void clear() + { + delegatedClear(); + } + + public boolean containsKey(Object key) + { + return delegatedContainsKey(key); + } + + public boolean containsValue(Object value) + { + return delegatedContainsValue(value); + } + + public Set<Entry<K, V>> entrySet() + { + return delegatedEntrySet(); + } + + public V get(Object key) + { + return delegatedGet(key); + } + + public boolean isEmpty() + { + return delegatedIsEmpty(); + } + + public Set<K> keySet() + { + return delegatedKeySet(); + } + + public V put(K key, V value) + { + return delegatedPut(key, value); + } + + public void putAll(Map<? extends K, ? extends V> t) + { + delegatedPutAll(t); + } + + public V remove(Object key) + { + return delegatedRemove(key); + } + + public int size() + { + return delegatedSize(); + } + + public Collection<V> values() + { + return delegatedValues(); + } + + @Override + public boolean equals(Object obj) + { + return delegatedEquals(obj); + } + + @Override + public int hashCode() + { + return delegatedHashCode(); + } + + @Override + public String toString() + { + return delegatedToString(); + } + + protected void delegatedClear() + { + getDelegate().clear(); + } + + protected boolean delegatedContainsKey(Object key) + { + return getDelegate().containsKey(key); + } + + protected boolean delegatedContainsValue(Object value) + { + return getDelegate().containsValue(value); + } + + protected Set<Entry<K, V>> delegatedEntrySet() + { + return getDelegate().entrySet(); + } + + protected V delegatedGet(Object key) + { + return getDelegate().get(key); + } + + protected boolean delegatedIsEmpty() + { + return getDelegate().isEmpty(); + } + + protected Set<K> delegatedKeySet() + { + return getDelegate().keySet(); + } + + protected V delegatedPut(K key, V value) + { + return getDelegate().put(key, value); + } + + protected void delegatedPutAll(Map<? extends K, ? extends V> t) + { + getDelegate().putAll(t); + } + + protected V delegatedRemove(Object key) + { + return getDelegate().remove(key); + } + + protected int delegatedSize() + { + return getDelegate().size(); + } + + protected Collection<V> delegatedValues() + { + return getDelegate().values(); + } + + protected boolean delegatedEquals(Object obj) + { + return getDelegate().equals(obj); + } + + protected int delegatedHashCode() + { + return getDelegate().hashCode(); + } + + protected String delegatedToString() + { + return getDelegate().toString(); + } + + protected abstract Map<K, V> getDelegate(); +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/registry/AbstractCachingRegistry.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/registry/AbstractCachingRegistry.java new file mode 100644 index 0000000000..c7b116fc24 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/registry/AbstractCachingRegistry.java @@ -0,0 +1,162 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.util.registry; + +import org.eclipse.net4j.util.registry.IRegistry.Listener.EventType; +import org.eclipse.net4j.util.registry.IRegistryElement.Descriptor; + +import java.util.HashSet; +import java.util.Map; +import java.util.Set; + +/** + * TODO Check if all methods of DelegatingRegistry still do what they should. + * TODO Remove DelegatingRegistry? + * + * @author Eike Stepper + */ +public abstract class AbstractCachingRegistry<ID, E extends IRegistryElement<ID>> extends + DelegatingRegistry<ID, E> +{ + public AbstractCachingRegistry(IRegistry<ID, E> delegate) + { + this(delegate, DEFAULT_RESOLVING); + } + + public AbstractCachingRegistry(IRegistry<ID, E> delegate, boolean resolving) + { + super(delegate, resolving); + } + + @Override + public synchronized void register(E element) + { + ID id = element.getID(); + E delegatedElement = super.lookup(id, false); + E oldElement = getCache().put(id, element); + + if (oldElement == null) + { + if (delegatedElement != null) + { + // Unhidden delegated element now becomes hidden + fireElementDeregistering(delegatedElement); + } + } + else + { + fireElementDeregistering(oldElement); + oldElement.dispose(); + } + + fireElementRegistered(element); + } + + @Override + public synchronized void deregister(ID id) + { + E delegatedElement = super.lookup(id, false); + E element = getCache().remove(id); + + if (element != null) + { + fireElementDeregistering(element); + element.dispose(); + + if (delegatedElement != null) + { + // Hidden delegated element now becomes unhidden + fireElementRegistered(delegatedElement); + } + } + } + + /** + * Synchronized to support {@link #resolveElement(IRegistryElement)} + */ + @Override + public synchronized E lookup(ID id, boolean resolve) + { + E element = getCache().get(id); + if (element == null) + { + if (resolve) + { + return resolveDelegatedElement(id); + } + else + { + return super.lookup(id, false); + } + } + + if (resolve) + { + element = resolveElement(element); + } + + return element; + } + + @Override + public Set<ID> getElementIDs() + { + Set<ID> ids = new HashSet(); + ids.addAll(super.getElementIDs()); + ids.addAll(getCache().keySet()); + return ids; + } + + @Override + public synchronized void dispose() + { + for (E element : getCache().values()) + { + fireElementDeregistering(element); + element.dispose(); + } + + getCache().clear(); + super.dispose(); + } + + @Override + protected void replaceElement(ID id, E element) + { + getCache().put(id, element); + } + + @Override + protected synchronized void handleDelegateEvent(EventType eventType, E element) + { + if (!getCache().containsKey(element.getID())) + { + fireRegistryEvent(eventType, element); + } + } + + protected E resolveDelegatedElement(ID id) + { + synchronized (getDelegate()) + { + boolean wasUnresolved = super.lookup(id, false) instanceof Descriptor; + E e = super.lookup(id, true); + if (wasUnresolved) + { + fireElementResolved(e); + } + + return e; + } + } + + protected abstract Map<ID, E> getCache(); +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/registry/AbstractMappingRegistry.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/registry/AbstractMappingRegistry.java new file mode 100644 index 0000000000..e9c4fd97e6 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/registry/AbstractMappingRegistry.java @@ -0,0 +1,91 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.util.registry; + +import java.util.Map; +import java.util.Set; + +/** + * @author Eike Stepper + */ +public abstract class AbstractMappingRegistry<ID, E extends IRegistryElement<ID>> extends + AbstractRegistry<ID, E> +{ + public AbstractMappingRegistry() + { + this(DEFAULT_RESOLVING); + } + + public AbstractMappingRegistry(boolean resolving) + { + super(resolving); + } + + public synchronized void register(E element) + { + E oldElement = getMap().put(element.getID(), element); + if (oldElement != null) + { + fireElementDeregistering(oldElement); + oldElement.dispose(); + } + + fireElementRegistered(element); + } + + public synchronized void deregister(ID id) + { + E element = getMap().remove(id); + if (element != null) + { + fireElementDeregistering(element); + element.dispose(); + } + } + + /** + * Synchronized to support {@link #resolveElement(IRegistryElement)} + */ + public synchronized E lookup(ID id, boolean resolve) + { + E element = getMap().get(id); + if (resolve) + { + element = resolveElement(element); + } + + return element; + } + + public Set<ID> getElementIDs() + { + return getMap().keySet(); + } + + @Override + public synchronized void dispose() + { + for (E element : getMap().values()) + { + element.dispose(); + } + + getMap().clear(); + } + + @Override + protected void replaceElement(ID id, E element) + { + getMap().put(id, element); + } + + protected abstract Map<ID, E> getMap(); +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/registry/AbstractRegistry.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/registry/AbstractRegistry.java new file mode 100644 index 0000000000..3bba7218ef --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/registry/AbstractRegistry.java @@ -0,0 +1,163 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.util.registry; + +import org.eclipse.net4j.util.registry.IRegistry.Listener.EventType; +import org.eclipse.net4j.util.registry.IRegistryElement.Descriptor; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.Queue; +import java.util.Set; +import java.util.concurrent.ConcurrentLinkedQueue; + +/** + * @author Eike Stepper + */ +public abstract class AbstractRegistry<ID, E extends IRegistryElement<ID>> implements + IRegistry<ID, E> +{ + public static final boolean DEFAULT_RESOLVING = true; + + private boolean resolving; + + /** + * Don't initialize lazily to circumvent synchronization! + */ + private Queue<Listener> listeners = new ConcurrentLinkedQueue(); + + protected AbstractRegistry() + { + this(DEFAULT_RESOLVING); + } + + public AbstractRegistry(boolean resolving) + { + setResolving(resolving); + } + + public void setResolving(boolean resolving) + { + this.resolving = resolving; + } + + public final boolean isResolving() + { + return resolving; + } + + public final boolean isResolved(ID id) + { + E element = lookup(id, false); + return element != null && !(element instanceof Descriptor); + } + + public final boolean isRegistered(ID id) + { + return lookup(id, false) != null; + } + + public int size() + { + return getElementIDs().size(); + } + + public final E lookup(ID id) + { + return lookup(id, isResolving()); + } + + public final Collection<E> getElements() + { + return getElements(isResolving()); + } + + /** + * Synchronized to support {@link #resolveElement(IRegistryElement)} + */ + public final synchronized Collection<E> getElements(boolean resolve) + { + Set<ID> elementKeys = getElementIDs(); + List<E> elements = new ArrayList<E>(elementKeys.size()); + for (ID id : elementKeys) + { + elements.add(lookup(id, resolve)); + } + + return elements; + } + + public void addListener(IRegistry.Listener<ID, E> listener) + { + listeners.add(listener); + } + + public void removeListener(IRegistry.Listener<ID, E> listener) + { + listeners.remove(listener); + } + + public void dispose() + { + listeners.clear(); + } + + protected void fireRegistryEvent(EventType eventType, E element) + { + for (Listener listener : listeners) + { + try + { + listener.notifyRegistryEvent(this, eventType, element); + } + catch (Exception ex) + { + ex.printStackTrace(); + } + } + } + + protected void fireElementRegistered(E element) + { + fireRegistryEvent(EventType.REGISTERED, element); + } + + protected void fireElementDeregistering(E element) + { + fireRegistryEvent(EventType.DEREGISTERING, element); + } + + protected void fireElementResolved(E element) + { + fireRegistryEvent(EventType.RESOLVED, element); + } + + /** + * Calling thread must already synchronize on this {@link IRegistry}! + */ + protected E resolveElement(E element) + { + if (element instanceof Descriptor) + { + element = (E)((Descriptor)element).resolve(); + replaceElement(element.getID(), element); + fireElementResolved(element); + } + + return element; + } + + /** + * Calling thread must already synchronize on this {@link IRegistry}! + */ + protected abstract void replaceElement(ID id, E element); +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/registry/DelegatingRegistry.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/registry/DelegatingRegistry.java new file mode 100644 index 0000000000..9676232c58 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/registry/DelegatingRegistry.java @@ -0,0 +1,115 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.util.registry; + +import org.eclipse.net4j.util.registry.IRegistry.Listener.EventType; + +import java.util.Set; + +/** + * @author Eike Stepper + */ +public class DelegatingRegistry<ID, E extends IRegistryElement<ID>> extends AbstractRegistry<ID, E> +{ + private IRegistry<ID, E> delegate; + + private Listener<ID, E> delegateListener = new Listener<ID, E>() + { + public void notifyRegistryEvent(IRegistry<ID, E> registry, EventType eventType, E element) + { + handleDelegateEvent(eventType, element); + } + }; + + public DelegatingRegistry(IRegistry<ID, E> delegate) + { + this(delegate, DEFAULT_RESOLVING); + } + + public DelegatingRegistry(IRegistry<ID, E> delegate, boolean resolving) + { + super(resolving); + this.delegate = delegate; + delegate.addListener(delegateListener); + } + + public IRegistry<ID, E> getDelegate() + { + return delegate; + } + + public void register(E element) + { + delegatedRegister(element); + } + + public void deregister(ID id) + { + delegatedDeregister(id); + } + + public E lookup(ID id, boolean resolve) + { + return delegatedLookup(id, resolve); + } + + public Set<ID> getElementIDs() + { + return delegatedGetElementIDs(); + } + + @Override + public void dispose() + { + delegate.removeListener(delegateListener); + } + + protected void delegatedRegister(E element) + { + delegate.register(element); + } + + protected void delegatedDeregister(ID id) + { + delegate.deregister(id); + } + + protected E delegatedLookup(ID id, boolean resolve) + { + return delegate.lookup(id, resolve); + } + + protected Set<ID> delegatedGetElementIDs() + { + return delegate.getElementIDs(); + } + + protected int delegatedSize() + { + return delegate.size(); + } + + protected void delegatedDispose() + { + delegate.dispose(); + } + + protected void handleDelegateEvent(EventType eventType, E element) + { + fireRegistryEvent(eventType, element); + } + + @Override + protected void replaceElement(ID id, E element) + { + // Do nothing + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/registry/HashCacheRegistry.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/registry/HashCacheRegistry.java new file mode 100644 index 0000000000..29f80d4cbc --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/registry/HashCacheRegistry.java @@ -0,0 +1,45 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.util.registry; + +import java.util.HashMap; +import java.util.Map; + +/** + * @author Eike Stepper + */ +public class HashCacheRegistry<ID, E extends IRegistryElement<ID>> extends + AbstractCachingRegistry<ID, E> +{ + private Map<ID, E> cache; + + public HashCacheRegistry(IRegistry<ID, E> delegate) + { + this(delegate, DEFAULT_RESOLVING); + } + + public HashCacheRegistry(IRegistry<ID, E> delegate, boolean resolving) + { + super(delegate, resolving); + cache = createCache(); + } + + @Override + protected Map<ID, E> getCache() + { + return cache; + } + + protected Map<ID, E> createCache() + { + return new HashMap(); + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/registry/HashMapRegistry.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/registry/HashMapRegistry.java new file mode 100644 index 0000000000..83b75fdaf3 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/registry/HashMapRegistry.java @@ -0,0 +1,45 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.util.registry; + +import java.util.HashMap; +import java.util.Map; + +/** + * @author Eike Stepper + */ +public class HashMapRegistry<ID, E extends IRegistryElement<ID>> extends + AbstractMappingRegistry<ID, E> +{ + private Map<ID, E> map; + + public HashMapRegistry() + { + this(DEFAULT_RESOLVING); + } + + public HashMapRegistry(boolean resolving) + { + super(resolving); + this.map = createMap(); + } + + @Override + protected Map<ID, E> getMap() + { + return map; + } + + protected Map<ID, E> createMap() + { + return new HashMap(); + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/registry/IRegistry.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/registry/IRegistry.java new file mode 100644 index 0000000000..86a36a9ef5 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/registry/IRegistry.java @@ -0,0 +1,70 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.util.registry; + +import java.util.Collection; +import java.util.Set; + +/** + * Implementation note: {@link Object#equals(Object)} and + * {@link Object#hashCode()} are based on pointer equality. + * <p> + * + * @author Eike Stepper + */ +public interface IRegistry<ID, E extends IRegistryElement<ID>> +{ + public void setResolving(boolean resolving); + + public boolean isResolving(); + + public boolean isResolved(ID id); + + public boolean isRegistered(ID id); + + public int size(); + + public void register(E element); + + public void deregister(ID id); + + public E lookup(ID id); + + public E lookup(ID id, boolean resolve); + + public Set<ID> getElementIDs(); + + public Collection<E> getElements(); + + public Collection<E> getElements(boolean resolve); + + public void addListener(Listener<ID, E> listener); + + public void removeListener(Listener<ID, E> listener); + + public void dispose(); + + /** + * @author Eike Stepper + */ + public interface Listener<ID, E extends IRegistryElement<ID>> + { + public void notifyRegistryEvent(IRegistry<ID, E> registry, EventType eventType, E element); + + /** + * @author Eike Stepper + */ + public enum EventType + { + REGISTERED, DEREGISTERING, RESOLVED + } + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/registry/IRegistryElement.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/registry/IRegistryElement.java new file mode 100644 index 0000000000..b25f11e9b6 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/registry/IRegistryElement.java @@ -0,0 +1,19 @@ +package org.eclipse.net4j.util.registry; + +/** + * @author Eike Stepper + */ +public interface IRegistryElement<ID> +{ + public ID getID(); + + public void dispose(); + + /** + * @author Eike Stepper + */ + public interface Descriptor<ID> extends IRegistryElement<ID> + { + public IRegistryElement<ID> resolve(); + } +}
\ No newline at end of file diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/stream/BufferInputStream.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/stream/BufferInputStream.java new file mode 100644 index 0000000000..4208f91e91 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/stream/BufferInputStream.java @@ -0,0 +1,142 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.util.stream; + +import org.eclipse.net4j.transport.Buffer; +import org.eclipse.net4j.transport.BufferHandler; + +import java.io.IOException; +import java.io.InputStream; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; + +/** + * @author Eike Stepper + */ +public class BufferInputStream extends InputStream implements BufferHandler +{ + public static final long NO_TIMEOUT = -1; + + public static final long DEFAULT_MILLIS_BEFORE_TIMEOUT = NO_TIMEOUT; + + public static final long DEFAULT_MILLIS_INTERRUPT_CHECK = 100; + + private BlockingQueue<Buffer> buffers = new LinkedBlockingQueue<Buffer>(); + + private Buffer currentBuffer; + + private boolean eos; + + public BufferInputStream() + { + } + + public void handleBuffer(Buffer buffer) + { + buffers.add(buffer); + } + + @Override + public int read() throws IOException + { + if (eos && currentBuffer == null) + { + // End of sequence + return -1; + } + + if (!ensureBuffer()) + { + // Timeout or interrupt + return -1; + } + + int result = currentBuffer.getByteBuffer().get() - Byte.MIN_VALUE; + if (!currentBuffer.getByteBuffer().hasRemaining()) + { + currentBuffer.release(); + currentBuffer = null; + } + + return result; + } + + @Override + public void close() throws IOException + { + buffers = null; + currentBuffer = null; + super.close(); + } + + @Override + public String toString() + { + return "BufferInputStream"; + } + + protected boolean ensureBuffer() throws IOException + { + try + { + final long check = getMillisInterruptCheck(); + final long timeout = getMillisBeforeTimeout(); + + try + { + if (timeout == NO_TIMEOUT) + { + while (currentBuffer == null) + { + currentBuffer = buffers.poll(check, TimeUnit.MILLISECONDS); + } + } + else + { + final long stop = System.currentTimeMillis() + timeout; + while (currentBuffer == null) + { + final long remaining = stop - System.currentTimeMillis(); + if (remaining <= 0) + { + return false; + } + + currentBuffer = buffers.poll(Math.min(remaining, check), TimeUnit.MILLISECONDS); + } + } + } + catch (InterruptedException ex) + { + throw new IOException("Interrupted"); + } + + eos = currentBuffer.isEOS(); + } + catch (RuntimeException ex) + { + // TODO Remove + ex.printStackTrace(); + } + return true; + } + + public long getMillisBeforeTimeout() + { + return DEFAULT_MILLIS_BEFORE_TIMEOUT; + } + + public long getMillisInterruptCheck() + { + return DEFAULT_MILLIS_INTERRUPT_CHECK; + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/stream/BufferOutputStream.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/stream/BufferOutputStream.java new file mode 100644 index 0000000000..ab67ac766d --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/stream/BufferOutputStream.java @@ -0,0 +1,138 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.util.stream; + +import org.eclipse.net4j.transport.Buffer; +import org.eclipse.net4j.transport.BufferHandler; +import org.eclipse.net4j.transport.BufferProvider; +import org.eclipse.net4j.util.lifecycle.LifecycleUtil; + +import java.io.IOException; +import java.io.OutputStream; +import java.nio.ByteBuffer; + +/** + * @author Eike Stepper + */ +public class BufferOutputStream extends OutputStream +{ + public static final boolean DEFAULT_PROPAGATE_CLOSE = false; + + private BufferHandler bufferHandler; + + private BufferProvider bufferProvider; + + private Buffer currentBuffer; + + private short channelID; + + public BufferOutputStream(BufferHandler bufferHandler, BufferProvider bufferProvider, + short channelID) + { + if (bufferHandler == null) + { + throw new IllegalArgumentException("bufferHandler == null"); + } + + if (bufferProvider == null) + { + throw new IllegalArgumentException("bufferProvider == null"); + } + + this.bufferHandler = bufferHandler; + this.bufferProvider = bufferProvider; + this.channelID = channelID; + } + + public BufferOutputStream(BufferHandler bufferHandler, short channelID) + { + this(bufferHandler, extractBufferProvider(bufferHandler), channelID); + } + + @Override + public void write(int b) throws IOException + { + ensureBuffer(); + ByteBuffer buffer = currentBuffer.getByteBuffer(); + buffer.put((byte)(b + Byte.MIN_VALUE)); + + if (!buffer.hasRemaining()) + { + flush(); + } + } + + @Override + public void flush() throws IOException + { + if (currentBuffer != null) + { + bufferHandler.handleBuffer(currentBuffer); + currentBuffer = null; + } + } + + public void flushWithEOS() throws IOException + { + ensureBuffer(); + currentBuffer.setEOS(true); + flush(); + } + + @Override + public void close() throws IOException + { + try + { + if (isPropagateClose()) + { + LifecycleUtil.deactivate(bufferHandler); + } + } + finally + { + bufferHandler = null; + bufferProvider = null; + currentBuffer = null; + super.close(); + } + } + + @Override + public String toString() + { + return "BufferOutputStream"; + } + + protected void ensureBuffer() + { + if (currentBuffer == null) + { + currentBuffer = bufferProvider.provideBuffer(); + currentBuffer.startPutting(channelID); + } + } + + protected boolean isPropagateClose() + { + return DEFAULT_PROPAGATE_CLOSE; + } + + private static BufferProvider extractBufferProvider(BufferHandler bufferHandler) + { + if (bufferHandler instanceof BufferProvider) + { + return (BufferProvider)bufferHandler; + } + + throw new IllegalArgumentException("Buffer handler unable to provide buffers"); + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/stream/ChannelInputStream.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/stream/ChannelInputStream.java new file mode 100644 index 0000000000..81cfb487f8 --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/stream/ChannelInputStream.java @@ -0,0 +1,69 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.util.stream; + +import org.eclipse.net4j.transport.Channel; + +/** + * @author Eike Stepper + */ +public class ChannelInputStream extends BufferInputStream +{ + private Channel channel; + + private long millisBeforeTimeout = DEFAULT_MILLIS_BEFORE_TIMEOUT; + + private long millisInterruptCheck = DEFAULT_MILLIS_INTERRUPT_CHECK; + + public ChannelInputStream(Channel channel) + { + this(channel, DEFAULT_MILLIS_BEFORE_TIMEOUT); + } + + public ChannelInputStream(Channel channel, long millisBeforeTimeout) + { + this.channel = channel; + channel.setReceiveHandler(this); + this.millisBeforeTimeout = millisBeforeTimeout; + millisInterruptCheck = DEFAULT_MILLIS_INTERRUPT_CHECK; + } + + public Channel getChannel() + { + return channel; + } + + public long getMillisBeforeTimeout() + { + return millisBeforeTimeout; + } + + public void setMillisBeforeTimeout(long millisBeforeTimeout) + { + this.millisBeforeTimeout = millisBeforeTimeout; + } + + public long getMillisInterruptCheck() + { + return millisInterruptCheck; + } + + public void setMillisInterruptCheck(long millisInterruptCheck) + { + this.millisInterruptCheck = millisInterruptCheck; + } + + @Override + public String toString() + { + return "ChannelInputStream[" + channel + "]"; + } +} diff --git a/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/stream/ChannelOutputStream.java b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/stream/ChannelOutputStream.java new file mode 100644 index 0000000000..2314836ffa --- /dev/null +++ b/plugins/org.eclipse.net4j/src/org/eclipse/net4j/util/stream/ChannelOutputStream.java @@ -0,0 +1,30 @@ +/*************************************************************************** + * Copyright (c) 2004, 2005, 2006 Eike Stepper, Germany. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Eike Stepper - initial API and implementation + **************************************************************************/ +package org.eclipse.net4j.util.stream; + +import org.eclipse.net4j.transport.BufferProvider; +import org.eclipse.net4j.transport.Channel; + +/** + * @author Eike Stepper + */ +public class ChannelOutputStream extends BufferOutputStream +{ + public ChannelOutputStream(Channel channel) + { + super(channel, channel.getChannelID()); + } + + public ChannelOutputStream(Channel channel, BufferProvider bufferProvider) + { + super(channel, bufferProvider, channel.getChannelID()); + } +} |
