TCF Agent: Implemented Streams service.

Added Streams service tests.
Improved comments in multiple files.

10 files changed, 860 insertions, 25 deletions

diff --git a/plugins/org.eclipse.tm.tcf.debug/src/org/eclipse/tm/internal/tcf/debug/tests/TCFTestSuite.java b/plugins/org.eclipse.tm.tcf.debug/src/org/eclipse/tm/internal/tcf/debug/tests/TCFTestSuite.java
index 1dc96eb3b..047396044 100644
--- a/plugins/org.eclipse.tm.tcf.debug/src/org/eclipse/tm/internal/tcf/debug/tests/TCFTestSuite.java
+++ b/plugins/org.eclipse.tm.tcf.debug/src/org/eclipse/tm/internal/tcf/debug/tests/TCFTestSuite.java
@@ -84,6 +84,14 @@ public class TCFTestSuite {
         });
         pending_tests.add(new Runnable() {
             public void run() {
+                listener.progress("Running Streams Test...", ++count_done, count_total);
+                for (IChannel channel : channels) {
+                    active_tests.put(new TestStreams(TCFTestSuite.this, channel), channel);
+                }
+            }
+        });
+        pending_tests.add(new Runnable() {
+            public void run() {
                 int i = 0;
                 listener.progress("Running Run Control Test...", ++count_done, count_total);
                 for (IChannel channel : channels) {
diff --git a/plugins/org.eclipse.tm.tcf.debug/src/org/eclipse/tm/internal/tcf/debug/tests/TestExpressions.java b/plugins/org.eclipse.tm.tcf.debug/src/org/eclipse/tm/internal/tcf/debug/tests/TestExpressions.java
index e1e658be9..d25c32e12 100644
--- a/plugins/org.eclipse.tm.tcf.debug/src/org/eclipse/tm/internal/tcf/debug/tests/TestExpressions.java
+++ b/plugins/org.eclipse.tm.tcf.debug/src/org/eclipse/tm/internal/tcf/debug/tests/TestExpressions.java
@@ -410,7 +410,10 @@ class TestExpressions implements ITCFTest,
             String s = (String)status.get(IBreakpoints.STATUS_ERROR);
             if (s != null) exit(new Exception(s));
             Collection<Map<String,Object>> list = (Collection<Map<String,Object>>)status.get(IBreakpoints.STATUS_INSTANCES);
-            if (list == null) exit(new Exception("Invalis BP status"));
+            if (list == null) {
+                exit(new Exception("Invalis BP status"));
+                return;
+            }
             boolean ok = false;
             for (Map<String,Object> map : list) {
                 String ctx = (String)map.get(IBreakpoints.INSTANCE_CONTEXT);
diff --git a/plugins/org.eclipse.tm.tcf.debug/src/org/eclipse/tm/internal/tcf/debug/tests/TestStreams.java b/plugins/org.eclipse.tm.tcf.debug/src/org/eclipse/tm/internal/tcf/debug/tests/TestStreams.java
new file mode 100644
index 000000000..7cddd7073
--- /dev/null
+++ b/plugins/org.eclipse.tm.tcf.debug/src/org/eclipse/tm/internal/tcf/debug/tests/TestStreams.java
@@ -0,0 +1,242 @@
+package org.eclipse.tm.internal.tcf.debug.tests;
+
+import java.util.HashSet;
+import java.util.Random;
+
+import org.eclipse.tm.tcf.protocol.IChannel;
+import org.eclipse.tm.tcf.protocol.IToken;
+import org.eclipse.tm.tcf.services.IDiagnostics;
+import org.eclipse.tm.tcf.services.IStreams;
+
+class TestStreams implements ITCFTest, IStreams.StreamsListener {
+
+    private final TCFTestSuite test_suite;
+    private final IDiagnostics diag;
+    private final IStreams streams;
+    private final Random rnd = new Random();
+    private final HashSet<String> stream_ids = new HashSet<String>();
+    
+    private String inp_id;
+    private String out_id;
+    
+    private int test_count;
+
+    TestStreams(TCFTestSuite test_suite, IChannel channel) {
+        this.test_suite = test_suite;
+        diag = channel.getRemoteService(IDiagnostics.class);
+        streams = channel.getRemoteService(IStreams.class);
+    }
+    
+    public void start() {
+        if (diag == null ||streams == null) {
+            test_suite.done(this, null);
+        }
+        else {
+            subsrcibe();
+        }
+    }
+    
+    private void subsrcibe() {
+        streams.subscribe(IDiagnostics.NAME, this, new IStreams.DoneSubscribe() {
+
+            public void doneSubscribe(IToken token, Exception error) {
+                if (error != null) {
+                    exit(error);
+                }
+                else {
+                    createStream();
+                }
+            }
+        });
+    }
+    
+    private void createStream() {
+        diag.createTestStreams(1153, 947, new IDiagnostics.DoneCreateTestStreams() {
+
+            public void doneCreateTestStreams(IToken token, Throwable error, String inp_id, String out_id) {
+                if (error != null) {
+                    exit(error);
+                }
+                else {
+                    TestStreams.this.inp_id = inp_id;
+                    TestStreams.this.out_id = out_id;
+                    for (String id : stream_ids) {
+                        if (id.equals(inp_id)) continue;
+                        if (id.equals(out_id)) continue;
+                        streams.disconnect(id, new IStreams.DoneDisconnect() {
+    
+                            public void doneDisconnect(IToken token, Exception error) {
+                                if (error != null) {
+                                    exit(error);
+                                }
+                            }
+                        });
+                    }
+                    testReadWrite();
+                }
+            }
+        });
+    }
+    
+    private void testReadWrite() {
+        final byte[] data_out = new byte[rnd.nextInt(10000) + 1000];
+        new Random().nextBytes(data_out);
+        final HashSet<IToken> cmds = new HashSet<IToken>();
+        IStreams.DoneRead done_read = new IStreams.DoneRead() {
+            
+            private int offs = 0;
+            private boolean eos;
+
+            public void doneRead(IToken token, Exception error, int lost_size, byte[] data, boolean eos) {
+                cmds.remove(token);
+                if (error != null) {
+                    if (!this.eos) {
+                        exit(error);
+                        return;
+                    }
+                }
+                else if (lost_size != 0) {
+                    exit(new Exception("Streams service: unexpected data loss"));
+                    return;
+                }
+                else {
+                    if (this.eos) {
+                        if (!eos || data != null && data.length > 0) {
+                            exit(new Exception("Streams service: unexpected successful read after EOS"));
+                        }
+                    }
+                    else {
+                        if (data != null) {
+                            if (offs + data.length > data_out.length) {
+                                exit(new Exception("Streams service: read returns more data then expected"));
+                                return;
+                            }
+                            for (int n = 0; n < data.length; n++) {
+                                if (data[n] != data_out[offs]) {
+                                    exit(new Exception("Streams service: data error: " + data[n] + " != " + data_out[offs]));
+                                    return;
+                                }
+                                offs++;
+                            }
+                        }
+                        if (eos) {
+                            if (offs != data_out.length) {
+                                exit(new Exception("Streams service: unexpected EOS"));
+                                return;
+                            }
+                            this.eos = true;
+                        }
+                        if (!this.eos && cmds.size() < 8) {
+                            cmds.add(streams.read(out_id, 241, this));
+                        }
+                    }
+                }
+                if (cmds.isEmpty()) disposeStreams();
+            }
+        };
+        cmds.add(streams.read(out_id, 223, done_read));
+        cmds.add(streams.read(out_id, 227, done_read));
+        cmds.add(streams.read(out_id, 229, done_read));
+        cmds.add(streams.read(out_id, 233, done_read));
+        
+        IStreams.DoneWrite done_write = new IStreams.DoneWrite() {
+
+            public void doneWrite(IToken token, Exception error) {
+                if (error != null) exit(error);
+            }
+        };
+        int offs = 0;
+        while (offs < data_out.length) {
+            int size = rnd.nextInt(400);
+            if (size > data_out.length - offs) size = data_out.length - offs;
+            streams.write(inp_id, data_out, offs, size, done_write);
+            offs += size;
+        }
+        streams.eos(inp_id, new IStreams.DoneEOS() {
+
+            public void doneEOS(IToken token, Exception error) {
+                if (error != null) exit(error);
+            }
+        });
+    }
+    
+    private void disposeStreams() {
+        final HashSet<IToken> cmds = new HashSet<IToken>();
+        IStreams.DoneDisconnect done_disconnect = new IStreams.DoneDisconnect() {
+
+            public void doneDisconnect(IToken token, Exception error) {
+                if (error != null) {
+                    exit(error);
+                }
+                else {
+                    cmds.remove(token);
+                    if (cmds.isEmpty()) unsubscribe();
+                }
+            }
+        };
+        IDiagnostics.DoneDisposeTestStream done_dispose = new IDiagnostics.DoneDisposeTestStream() {
+
+            public void doneDisposeTestStream(IToken token, Throwable error) {
+                if (error != null) {
+                    exit(error);
+                }
+                else {
+                    cmds.remove(token);
+                    if (cmds.isEmpty()) unsubscribe();
+                }
+            }
+        };
+        cmds.add(streams.disconnect(inp_id, done_disconnect));
+        cmds.add(diag.disposeTestStream(inp_id, done_dispose));
+        cmds.add(diag.disposeTestStream(out_id, done_dispose));
+        cmds.add(streams.disconnect(out_id, done_disconnect));
+    }
+    
+    private void unsubscribe() {
+        streams.unsubscribe(IDiagnostics.NAME, this, new IStreams.DoneUnsubscribe() {
+
+            public void doneUnsubscribe(IToken token, Exception error) {
+                if (error != null || test_count >= 10) {
+                    exit(error);
+                }
+                else {
+                    test_count++;
+                    stream_ids.clear();
+                    inp_id = null;
+                    out_id = null;
+                    subsrcibe();
+                }
+            }
+        });
+    }
+
+    private void exit(Throwable x) {
+        if (!test_suite.isActive(this)) return;
+        test_suite.done(this, x);
+    }
+    
+    /************************** StreamsListener **************************/
+
+    public void created(String stream_type, String stream_id) {
+        if (!IDiagnostics.NAME.equals(stream_type)) exit(new Exception("Invalid stream type in Streams.created event"));
+        if (stream_ids.contains(stream_id)) exit(new Exception("Invalid stream ID in Streams.created event"));
+        stream_ids.add(stream_id);
+        if (inp_id != null) {
+            if (inp_id.equals(stream_id)) exit(new Exception("Invalid stream ID in Streams.created event"));
+            if (out_id.equals(stream_id)) exit(new Exception("Invalid stream ID in Streams.created event"));
+            streams.disconnect(stream_id, new IStreams.DoneDisconnect() {
+                
+                public void doneDisconnect(IToken token, Exception error) {
+                    if (error != null) {
+                        exit(error);
+                    }
+                }
+            });
+        }
+    }
+
+    public void disposed(String stream_type, String stream_id) {
+        if (!IDiagnostics.NAME.equals(stream_type)) exit(new Exception("Invalid stream type in Streams.disposed event"));
+        if (!stream_ids.remove(stream_id)) exit(new Exception("Invalid stream ID in Streams.disposed event"));
+    }
+}
diff --git a/plugins/org.eclipse.tm.tcf/src/org/eclipse/tm/internal/tcf/services/local/DiagnosticsService.java b/plugins/org.eclipse.tm.tcf/src/org/eclipse/tm/internal/tcf/services/local/DiagnosticsService.java
index 25f6eff68..79374c65c 100644
--- a/plugins/org.eclipse.tm.tcf/src/org/eclipse/tm/internal/tcf/services/local/DiagnosticsService.java
+++ b/plugins/org.eclipse.tm.tcf/src/org/eclipse/tm/internal/tcf/services/local/DiagnosticsService.java
@@ -101,4 +101,24 @@ public class DiagnosticsService implements IDiagnostics {
         });
         return token;
     }
+
+    public IToken createTestStreams(int inp_buf_size, int out_buf_size, final DoneCreateTestStreams done) {
+        final IToken token = new Token();
+        Protocol.invokeLater(new Runnable() {
+            public void run() {
+                done.doneCreateTestStreams(token, new Exception("Not implemented"), null, null);
+            }
+        });
+        return token;
+    }
+
+    public IToken disposeTestStream(String id, final DoneDisposeTestStream done) {
+        final IToken token = new Token();
+        Protocol.invokeLater(new Runnable() {
+            public void run() {
+                done.doneDisposeTestStream(token, new Exception("Invalid context"));
+            }
+        });
+        return token;
+    }
 }
diff --git a/plugins/org.eclipse.tm.tcf/src/org/eclipse/tm/internal/tcf/services/remote/DiagnosticsProxy.java b/plugins/org.eclipse.tm.tcf/src/org/eclipse/tm/internal/tcf/services/remote/DiagnosticsProxy.java
index 3e3525b2b..3369e5300 100644
--- a/plugins/org.eclipse.tm.tcf/src/org/eclipse/tm/internal/tcf/services/remote/DiagnosticsProxy.java
+++ b/plugins/org.eclipse.tm.tcf/src/org/eclipse/tm/internal/tcf/services/remote/DiagnosticsProxy.java
@@ -145,6 +145,36 @@ public class DiagnosticsProxy implements IDiagnostics {
         }.token;
     }
 
+    public IToken createTestStreams(int inp_buf_size, int out_buf_size, final DoneCreateTestStreams done) {
+        return new Command(channel, this, "createTestStreams", new Object[]{ inp_buf_size, out_buf_size }) {
+            @Override
+            public void done(Exception error, Object[] args) {
+                String inp_id = null;
+                String out_id = null;
+                if (error == null) {
+                    assert args.length == 3;
+                    error = toError(args[0]);
+                    inp_id = (String)args[1];
+                    out_id = (String)args[2];
+                }
+                done.doneCreateTestStreams(token, error, inp_id, out_id);
+            }
+        }.token;
+    }
+
+    public IToken disposeTestStream(String id, final DoneDisposeTestStream done) {
+        return new Command(channel, this, "disposeTestStream", new Object[]{ id }) {
+            @Override
+            public void done(Exception error, Object[] args) {
+                if (error == null) {
+                    assert args.length == 1;
+                    error = toError(args[0]);
+                }
+                done.doneDisposeTestStream(token, error);
+            }
+        }.token;
+    }
+
     @SuppressWarnings("unchecked")
     private String[] toStringArray(Object o) {
         if (o == null) return null;
diff --git a/plugins/org.eclipse.tm.tcf/src/org/eclipse/tm/internal/tcf/services/remote/StreamsProxy.java b/plugins/org.eclipse.tm.tcf/src/org/eclipse/tm/internal/tcf/services/remote/StreamsProxy.java
new file mode 100644
index 000000000..4dfd93424
--- /dev/null
+++ b/plugins/org.eclipse.tm.tcf/src/org/eclipse/tm/internal/tcf/services/remote/StreamsProxy.java
@@ -0,0 +1,152 @@
+/*******************************************************************************
+ * Copyright (c) 2009 Wind River Systems, Inc. and others.
+ * 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:
+ *     Wind River Systems - initial API and implementation
+ *******************************************************************************/
+package org.eclipse.tm.internal.tcf.services.remote;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.eclipse.tm.tcf.core.Base64;
+import org.eclipse.tm.tcf.core.Command;
+import org.eclipse.tm.tcf.protocol.IChannel;
+import org.eclipse.tm.tcf.protocol.IToken;
+import org.eclipse.tm.tcf.protocol.JSON;
+import org.eclipse.tm.tcf.services.IStreams;
+
+public class StreamsProxy implements IStreams {
+
+    private final IChannel channel;
+    private final Map<StreamsListener,IChannel.IEventListener> listeners =
+        new HashMap<StreamsListener,IChannel.IEventListener>();
+
+    public StreamsProxy(IChannel channel) {
+        this.channel = channel;
+    }
+
+    public IToken disconnect(String stream_id, final DoneDisconnect done) {
+        return new Command(channel, this, "disconnect", new Object[]{ stream_id }) {
+            @Override
+            public void done(Exception error, Object[] args) {
+                if (error == null) {
+                    assert args.length == 1;
+                    error = toError(args[0]);
+                }
+                done.doneDisconnect(token, error);
+            }
+        }.token;
+    }
+
+    public IToken eos(String stream_id, final DoneEOS done) {
+        return new Command(channel, this, "eos", new Object[]{ stream_id }) {
+            @Override
+            public void done(Exception error, Object[] args) {
+                if (error == null) {
+                    assert args.length == 1;
+                    error = toError(args[0]);
+                }
+                done.doneEOS(token, error);
+            }
+        }.token;
+    }
+
+    public IToken read(String stream_id, int size, final DoneRead done) {
+        return new Command(channel, this, "read", new Object[]{ stream_id, size }) {
+            @Override
+            public void done(Exception error, Object[] args) {
+                int lost_size = 0;
+                byte data[] = null;
+                boolean eos = false;
+                if (error == null) {
+                    assert args.length == 4;
+                    String str = (String)args[0];
+                    error = toError(args[1]);
+                    lost_size = ((Number)args[2]).intValue();
+                    eos = ((Boolean)args[3]).booleanValue();
+                    if (str != null) data = Base64.toByteArray(str.toCharArray());
+                }
+                done.doneRead(token, error, lost_size, data, eos);
+            }
+        }.token;
+    }
+
+    public IToken subscribe(String stream_type, final StreamsListener listener, final DoneSubscribe done) {
+        return new Command(channel, this, "subscribe", new Object[]{ stream_type }) {
+            @Override
+            public void done(Exception error, Object[] args) {
+                if (error == null) {
+                    assert args.length == 1;
+                    error = toError(args[0]);
+                }
+                if (error == null) {
+                    IChannel.IEventListener l = new IChannel.IEventListener() {
+
+                        public void event(String name, byte[] data) {
+                            try {
+                                Object[] args = JSON.parseSequence(data);
+                                if (name.equals("created")) {
+                                    assert args.length == 2;
+                                    listener.created((String)args[0], (String)args[1]);
+                                }
+                                else if (name.equals("disposed")) {
+                                    assert args.length == 2;
+                                    listener.disposed((String)args[0], (String)args[1]);
+                                }
+                                else {
+                                    throw new IOException("Streams service: unknown event: " + name);
+                                }
+                            }
+                            catch (Throwable x) {
+                                channel.terminate(x);
+                            }
+                        }
+                    };
+                    listeners.put(listener, l);
+                    channel.addEventListener(StreamsProxy.this, l);
+                }
+                done.doneSubscribe(token, error);
+            }
+        }.token;
+    }
+
+    public IToken unsubscribe(String stream_type, final StreamsListener listener, final DoneUnsubscribe done) {
+        return new Command(channel, this, "unsubscribe", new Object[]{ stream_type }) {
+            @Override
+            public void done(Exception error, Object[] args) {
+                if (error == null) {
+                    assert args.length == 1;
+                    error = toError(args[0]);
+                }
+                if (error == null) {
+                    IChannel.IEventListener l = listeners.remove(listener);
+                    if (l != null) channel.removeEventListener(StreamsProxy.this, l);
+                }
+                done.doneUnsubscribe(token, error);
+            }
+        }.token;
+    }
+
+    public IToken write(String stream_id, byte[] buf, int offset, int size, final DoneWrite done) {
+        return new Command(channel, this, "write", new Object[]{ stream_id, size, Base64.toBase64(buf, offset, size) }) {
+            @Override
+            public void done(Exception error, Object[] args) {
+                if (error == null) {
+                    assert args.length == 1;
+                    error = toError(args[0]);
+                }
+                done.doneWrite(token, error);
+            }
+        }.token;
+    }
+
+    public String getName() {
+        return NAME;
+    }
+}
diff --git a/plugins/org.eclipse.tm.tcf/src/org/eclipse/tm/tcf/services/IBreakpoints.java b/plugins/org.eclipse.tm.tcf/src/org/eclipse/tm/tcf/services/IBreakpoints.java
index c82a36b01..9e40654bb 100644
--- a/plugins/org.eclipse.tm.tcf/src/org/eclipse/tm/tcf/services/IBreakpoints.java
+++ b/plugins/org.eclipse.tm.tcf/src/org/eclipse/tm/tcf/services/IBreakpoints.java
@@ -126,6 +126,11 @@ public interface IBreakpoints extends IService {
      * Call back interface for breakpoint service commands.
      */
     interface DoneCommand {
+        /**
+         * Called when command is done.
+         * @param token - command handle.
+         * @param error - error object or null.
+         */
         void doneCommand(IToken token, Exception error);
     }
 
@@ -138,6 +143,7 @@ public interface IBreakpoints extends IService {
      * 
      * @param properties - array of breakpoints.
      * @param done - command result call back object.
+     * @return - pending command handle.
      */
     IToken set(Map<String,Object>[] properties, DoneCommand done);
 
@@ -145,6 +151,7 @@ public interface IBreakpoints extends IService {
      * Called when breakpoint is added into breakpoints table.
      * @param properties - breakpoint properties.
      * @param done - command result call back object.
+     * @return - pending command handle.
      */
     IToken add(Map<String,Object> properties, DoneCommand done);
 
@@ -152,6 +159,7 @@ public interface IBreakpoints extends IService {
      * Called when breakpoint properties are changed.
      * @param properties - breakpoint properties.
      * @param done - command result call back object.
+     * @return - pending command handle.
      */
     IToken change(Map<String,Object> properties, DoneCommand done);
 
@@ -159,6 +167,7 @@ public interface IBreakpoints extends IService {
      * Tell target to change (only) PROP_ENABLED breakpoint property to 'true'.
      * @param ids - array of enabled breakpoint identifiers.
      * @param done - command result call back object.
+     * @return - pending command handle.
      */
     IToken enable(String[] ids, DoneCommand done);
 
@@ -166,6 +175,7 @@ public interface IBreakpoints extends IService {
      * Tell target to change (only) PROP_ENABLED breakpoint property to 'false'.
      * @param ids - array of disabled breakpoint identifiers.
      * @param done - command result call back object.
+     * @return - pending command handle.
      */
     IToken disable(String[] ids, DoneCommand done);
 
@@ -173,16 +183,27 @@ public interface IBreakpoints extends IService {
      * Tell target to remove breakpoints.
      * @param id - unique breakpoint identifier.
      * @param done - command result call back object.
+     * @return - pending command handle.
      */
     IToken remove(String[] ids, DoneCommand done);
     
     /**
      * Upload IDs of breakpoints known to target agent.
      * @param done - command result call back object.
+     * @return - pending command handle.
      */
     IToken getIDs(DoneGetIDs done);
 
+    /**
+     * Call back interface for 'getIDs' command.
+     */
     interface DoneGetIDs {
+        /**
+         * Called when 'getIDs' command is done. 
+         * @param token - command handle.
+         * @param error - error object or null.
+         * @param ids - IDs of breakpoints known to target agent.
+         */
         void doneGetIDs(IToken token, Exception error, String[] ids);
     }
 
@@ -193,7 +214,16 @@ public interface IBreakpoints extends IService {
      */
     IToken getProperties(String id, DoneGetProperties done);
 
+    /**
+     * Call back interface for 'getProperties' command.
+     */
     interface DoneGetProperties {
+        /**
+         * Called when 'getProperties' command is done. 
+         * @param token - command handle.
+         * @param error - error object or null.
+         * @param properties - properties of the breakpoint.
+         */
         void doneGetProperties(IToken token, Exception error, Map<String,Object> properties);
     }
 
@@ -201,10 +231,20 @@ public interface IBreakpoints extends IService {
      * Upload status of given breakpoint from target agent.
      * @param id - unique breakpoint identifier.
      * @param done - command result call back object.
+     * @return - pending command handle.
      */
     IToken getStatus(String id, DoneGetStatus done);
 
+    /**
+     * Call back interface for 'getStatus' command.
+     */
     interface DoneGetStatus {
+        /**
+         * Called when 'getStatus' command is done. 
+         * @param token - command handle.
+         * @param error - error object or null.
+         * @param status - status of the breakpoint.
+         */
         void doneGetStatus(IToken token, Exception error, Map<String,Object> status);
     }
 
@@ -217,10 +257,20 @@ public interface IBreakpoints extends IService {
      * capabilities.
      * @param id - a context ID or null.
      * @param done - command result call back object.
+     * @return - pending command handle.
      */
     IToken getCapabilities(String id, DoneGetCapabilities done);
 
+    /**
+     * Call back interface for 'getCapabilities' command.
+     */
     interface DoneGetCapabilities {
+        /**
+         * Called when 'getCapabilities' command is done. 
+         * @param token - command handle.
+         * @param error - error object or null.
+         * @param capabilities - breakpoints service capabilities description.
+         */
         void doneGetCapabilities(IToken token, Exception error, Map<String,Object> capabilities);
     }
 
diff --git a/plugins/org.eclipse.tm.tcf/src/org/eclipse/tm/tcf/services/IDiagnostics.java b/plugins/org.eclipse.tm.tcf/src/org/eclipse/tm/tcf/services/IDiagnostics.java
index 075420a84..004cafe83 100644
--- a/plugins/org.eclipse.tm.tcf/src/org/eclipse/tm/tcf/services/IDiagnostics.java
+++ b/plugins/org.eclipse.tm.tcf/src/org/eclipse/tm/tcf/services/IDiagnostics.java
@@ -21,45 +21,182 @@ import org.eclipse.tm.tcf.protocol.IToken;
 
 public interface IDiagnostics extends IService {
 
-    public static final String NAME = "Diagnostics";
+    static final String NAME = "Diagnostics";
     
-    public IToken echo(String s, DoneEcho done);
+    /**
+     * 'echo' command result returns same string that was given as command argument.
+     * The command is used to test communication channel ability to transmit arbitrary strings in
+     * both directions.  
+     * @param s - any string.
+     * @param done - command result call back object.
+     * @return - pending command handle.
+     */
+    IToken echo(String s, DoneEcho done);
     
-    public interface DoneEcho {
-        public void doneEcho(IToken token, Throwable error, String s);
+    /**
+     * Call back interface for 'echo' command.
+     */
+    interface DoneEcho {
+        /**
+         * Called when 'echo' command is done.
+         * @param token - command handle.
+         * @param error - error object or null.
+         * @param s - same string as the command argument.
+         */
+        void doneEcho(IToken token, Throwable error, String s);
     }
 
-    public IToken getTestList(DoneGetTestList done);
+    /**
+     * Get list of test names that are implemented by the service.
+     * Clients can request remote peer to run a test from the list.
+     * When started, a test performs a predefined set actions.
+     * Nature of test actions is uniquely identified by test name.
+     * Exact description of test actions is a contract between client and remote peer,
+     * and it is not part of Diagnostics service specifications.
+     * Clients should not attempt to run a test if they don't recognize the test name.
+     * @param done - command result call back object.
+     * @return - pending command handle.
+     */
+    IToken getTestList(DoneGetTestList done);
     
-    public interface DoneGetTestList {
-        public void doneGetTestList(IToken token, Throwable error, String[] list);
+    /**
+     * Call back interface for 'getTestList' command.
+     */
+    interface DoneGetTestList {
+        /**
+         * Called when 'getTestList' command is done.
+         * @param token - command handle.
+         * @param error - error object or null.
+         * @param list - names of tests that are supported by the peer. 
+         */
+        void doneGetTestList(IToken token, Throwable error, String[] list);
     }
     
-    public IToken runTest(String s, DoneRunTest done);
+    /**
+     * Run a test. When started, a test performs a predefined set actions.
+     * Nature of test actions is uniquely identified by test name.
+     * Running test usually has associated execution context ID.
+     * Depending on the test, the ID can be used with services RunControl and/or Processes services to control
+     * test execution, and to obtain test results.
+     * @param name - test name
+     * @param done - command result call back object.
+     * @return - pending command handle.
+     */
+    IToken runTest(String name, DoneRunTest done);
     
-    public interface DoneRunTest {
-        public void doneRunTest(IToken token, Throwable error, String context_id);
+    /**
+     * Call back interface for 'runTest' command.
+     */
+    interface DoneRunTest {
+        /**
+         * Called when 'runTest' command is done.
+         * @param token - command handle.
+         * @param error - error object or null.
+         * @param context_id - test execution contest ID.
+         */
+        void doneRunTest(IToken token, Throwable error, String context_id);
     }
 
-    public IToken cancelTest(String context_id, DoneCancelTest done);
+    /**
+     * Cancel execution of a test.
+     * @param context_id - text execution context ID.
+     * @param done - command result call back object.
+     * @return - pending command handle.
+     */
+    IToken cancelTest(String context_id, DoneCancelTest done);
     
-    public interface DoneCancelTest {
-        public void doneCancelTest(IToken token, Throwable error);
+    /**
+     * Call back interface for 'cancelTest' command.
+     */
+    interface DoneCancelTest {
+        /**
+         * Called when 'cancelTest' command is done.
+         * @param token - command handle.
+         * @param error - error object or null.
+         */
+        void doneCancelTest(IToken token, Throwable error);
     }
     
-    public IToken getSymbol(String context_id, String symbol_name, DoneGetSymbol done);
+    /**
+     * Get information about a symbol in text execution context.
+     * @param context_id
+     * @param symbol_name
+     * @param done
+     * @return
+     */
+    IToken getSymbol(String context_id, String symbol_name, DoneGetSymbol done);
     
-    public interface DoneGetSymbol {
-        public void doneGetSymbol(IToken token, Throwable error, ISymbol symbol);
+    /**
+     * Call back interface for 'getSymbol' command.
+     */
+    interface DoneGetSymbol {
+        /**
+         * Called when 'getSymbol' command is done.
+         * @param token - command handle.
+         * @param error - error object or null.
+         * @param symbol
+         */
+        void doneGetSymbol(IToken token, Throwable error, ISymbol symbol);
     }
     
-    public interface ISymbol {
-        public String getSectionName();
-        public Number getValue();
-        public boolean isUndef();
-        public boolean isCommon();
-        public boolean isGlobal();
-        public boolean isLocal();
-        public boolean isAbs();
+    /**
+     * Interface to access result value of 'getSymbol' command.
+     */
+    interface ISymbol {
+        String getSectionName();
+        Number getValue();
+        boolean isUndef();
+        boolean isCommon();
+        boolean isGlobal();
+        boolean isLocal();
+        boolean isAbs();
+    }
+    
+    /**
+     * Create a pair of virtual streams, @see IStreams service.
+     * Remote ends of the streams are connected, so any data sent into 'inp' stream
+     * will become for available for reading from 'out' stream.
+     * The command is used for testing virtual streams.
+     * @param inp_buf_size - buffer size in bytes of the input stream.
+     * @param out_buf_size - buffer size in bytes of the output stream.
+     * @param done - command result call back object.
+     * @return - pending command handle.
+     */
+    IToken createTestStreams(int inp_buf_size, int out_buf_size, DoneCreateTestStreams done);
+    
+    /**
+     * Call back interface for 'createTestStreams' command.
+     */
+    interface DoneCreateTestStreams {
+        
+        /**
+         * Called when 'createTestStreams' command is done.
+         * @param token - command handle.
+         * @param error - error object or null.
+         * @param inp_id - the input stream ID.
+         * @param out_id - the output stream ID.
+         */
+        void doneCreateTestStreams(IToken token, Throwable error, String inp_id, String out_id);
+    }
+    
+    /**
+     * Dispose a virtual stream that was created by 'createTestStreams' command.
+     * @param id - the stream ID.
+     * @param done - command result call back object.
+     * @return - pending command handle.
+     */
+    IToken disposeTestStream(String id, DoneDisposeTestStream done);
+    
+    /**
+     * Call back interface for 'disposeTestStream' command.
+     */
+    interface DoneDisposeTestStream {
+        
+        /**
+         * Called when 'createTestStreams' command is done.
+         * @param token - command handle.
+         * @param error - error object or null.
+         */
+        void doneDisposeTestStream(IToken token, Throwable error);
     }
 }
diff --git a/plugins/org.eclipse.tm.tcf/src/org/eclipse/tm/tcf/services/IMemory.java b/plugins/org.eclipse.tm.tcf/src/org/eclipse/tm/tcf/services/IMemory.java
index 38af9e736..3304bc6b6 100644
--- a/plugins/org.eclipse.tm.tcf/src/org/eclipse/tm/tcf/services/IMemory.java
+++ b/plugins/org.eclipse.tm.tcf/src/org/eclipse/tm/tcf/services/IMemory.java
@@ -38,6 +38,7 @@ public interface IMemory extends IService {
      *   
      * @param id – context ID. 
      * @param done - call back interface called when operation is completed.
+     * @return - pending command handle.
      */
     IToken getContext(String id, DoneGetContext done);
 
@@ -67,6 +68,7 @@ public interface IMemory extends IService {
      * to retrieve top level of the hierarchy, or one of context IDs retrieved
      * by previous getChildren commands.
      * @param done - call back interface called when operation is completed.
+     * @return - pending command handle.
      */
     IToken getChildren(String parent_context_id, DoneGetChildren done);
 
diff --git a/plugins/org.eclipse.tm.tcf/src/org/eclipse/tm/tcf/services/IStreams.java b/plugins/org.eclipse.tm.tcf/src/org/eclipse/tm/tcf/services/IStreams.java
new file mode 100644
index 000000000..ded45744c
--- /dev/null
+++ b/plugins/org.eclipse.tm.tcf/src/org/eclipse/tm/tcf/services/IStreams.java
@@ -0,0 +1,191 @@
+/*******************************************************************************
+ * Copyright (c) 2009 Wind River Systems, Inc. and others.
+ * 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:
+ *     Wind River Systems - initial API and implementation
+ *******************************************************************************/
+package org.eclipse.tm.tcf.services;
+
+import org.eclipse.tm.tcf.protocol.IService;
+import org.eclipse.tm.tcf.protocol.IToken;
+
+/**
+ * Streams service is a generic interface to support streaming of data between host and remote agents.
+ * 
+ * The service supports:
+ *  1. Asynchronous overlapped data streaming: multiple 'read' or 'write' command can be issued at same time, both peers
+ *  can continue data processing concurrently with data transmission.
+ *  2. Multicast: multiple clients can receive data from same stream.
+ *  3. Subscription model: clients are required to expressed interest in particular streams by subscribing for the service.
+ *  4. Flow control: peers can throttle data flow of individual streams by delaying 'read' and 'write' commands.   
+ */
+public interface IStreams extends IService {
+    
+    /**
+     * Service name.
+     */
+    static final String NAME = "Streams";
+
+    /**
+     * Clients can implement StreamsListener interface to be notified
+     * when a stream is created or disposed. The interface is registered with 'subscribe' command. 
+     * 
+     * When new stream is created, client must decide if it is interested in that particular stream instance.
+     * If not interested, client should send 'disconnect' command to allow remote peer to free resources and bandwidth.
+     * If not disconnected, client is required to send 'read' commands as necessary to prevent stream buffer overflow.
+     */
+    interface StreamsListener {
+        
+        /**
+         * Called when a new stream is created. 
+         * @param stream_type - source type of the stream.
+         * @param stream_id - ID of the stream.
+         */
+        void created(String stream_type, String stream_id);
+        
+        /**
+         * Called when a stream is disposed. 
+         * @param stream_type - source type of the stream.
+         * @param stream_id - ID of the stream.
+         */
+        void disposed(String stream_type, String stream_id);
+    }
+    
+    /**
+     * Clients must subscribe for one or more stream types to be able to send or receive stream data.
+     * Subscribers receive notifications when a stream of given type is created or disposed.
+     * Subscribers are required to respond with 'read' or 'disconnect' commands as necessary.
+     * @param stream_type - the stream source type.
+     * @param listener - client implementation of StreamsListener interface.
+     * @param done - command result call back object.
+     * @return - pending command handle.
+     */
+    IToken subscribe(String stream_type, StreamsListener listener, DoneSubscribe done);
+    
+    /**
+     * Call back interface for 'subscribe' command.
+     */
+    interface DoneSubscribe {
+        void doneSubscribe(IToken token, Exception error);
+    }
+
+    /**
+     * Unsubscribe the client from given stream source type.
+     * @param stream_type - the stream source type.
+     * @param listener - client implementation of StreamsListener interface.
+     * @param done - command result call back object.
+     * @return - pending command handle.
+     */
+    IToken unsubscribe(String stream_type, StreamsListener listener, DoneUnsubscribe done);
+    
+    /**
+     * Call back interface for 'unsubscribe' command.
+     */
+    interface DoneUnsubscribe {
+        void doneUnsubscribe(IToken token, Exception error);
+    }
+
+    /**
+     * Read data from a stream. If stream buffer is empty, the command will wait until data is available.
+     * Remote peer will continue to process other commands while 'read' command is pending.
+     * Client can send more 'read' commands without waiting for the first command to complete.
+     * Doing that improves communication channel bandwidth utilization.
+     * Pending 'read' commands will be executed in same order as issued.
+     * Client can delay sending of 'read' command if it is not ready to receive more data,
+     * however, delaying for too long can cause stream buffer overflow and lost of data.
+     * @param stream_id - ID of the stream.
+     * @param size - max number of bytes to read.
+     * @param done - command result call back object.
+     * @return - pending command handle.
+     */
+    IToken read(String stream_id, int size, DoneRead done);
+    
+    /**
+     * Call back interface for 'read' command.
+     */
+    interface DoneRead {
+        /**
+         * Called when 'read' command is done.
+         * @param token - command handle.
+         * @param error - error object or null.
+         * @param lost_size - number of bytes that were lost because of buffer overflow.
+         * 'lost_size' -1 means unknown number of bytes were lost.
+         * if both 'lost_size' and 'data.length' are non-zero then lost bytes are considered
+         * located right before read bytes. 
+         * @param data - bytes read from the stream.
+         * @param eos - true if end of stream was reached.
+         */
+        void doneRead(IToken token, Exception error, int lost_size, byte[] data, boolean eos);
+    }
+
+    /**
+     * Write data to a stream. If stream buffer is full, the command will wait until space is available.
+     * Remote peer will continue to process other commands while 'write' command is pending.
+     * Client can send more 'write' commands without waiting for the first command to complete.
+     * Doing that improves communication channel bandwidth utilization.
+     * Pending 'write' commands will be executed in same order as issued.
+     * @param stream_id - ID of the stream.
+     * @param buf - buffer that contains stream data.
+     * @param offset - byte offset in the buffer.
+     * @param size - number of bytes to write.
+     * @param done - command result call back object.
+     * @return - pending command handle.
+     */
+    IToken write(String stream_id, byte[] buf, int offset, int size, DoneWrite done);
+    
+    /**
+     * Call back interface for 'write' command.
+     */
+    interface DoneWrite {
+        /**
+         * Called when 'write' command is done.
+         * @param token - command handle.
+         * @param error - error object or null.
+         */
+        void doneWrite(IToken token, Exception error);
+    }
+    
+    /**
+     * Send End Of Stream marker to a stream. No more writing to the stream is allowed after that. 
+     * @param stream_id - ID of the stream.
+     * @param done - command result call back object.
+     * @return - pending command handle.
+     */
+    IToken eos(String stream_id, DoneEOS done);
+    
+    /**
+     * Call back interface for 'eos' command.
+     */
+    interface DoneEOS {
+        /**
+         * Called when 'eos' command is done.
+         * @param token - command handle.
+         * @param error - error object or null.
+         */
+        void doneEOS(IToken token, Exception error);
+    }
+
+    /**
+     * Disconnect client from a stream.
+     * @param stream_id - ID of the stream.
+     * @param done - command result call back object.
+     * @return - pending command handle.
+     */
+    IToken disconnect(String stream_id, DoneDisconnect done);
+    
+    /**
+     * Call back interface for 'disconnect' command.
+     */
+    interface DoneDisconnect {
+        /**
+         * Called when 'disconnect' command is done.
+         * @param token - command handle.
+         * @param error - error object or null.
+         */
+        void doneDisconnect(IToken token, Exception error);
+    }
+}