Objekte im NIO unblocking mode versenden

[Teots]

Hallo!

Ich habe vor NIO zu nutzen. Dabei tut sich mir nun das Problem auf, dass ich keine Objekt serizalisieren und versenden kann. Dabei bekomme ich immer eine java.nio.channels.IllegalBlockingModeException. Das tritt laut Dokumentation immer dann auf, wenn der nichtblockierende Modus (configureBlocking(false)) verwendet wird. Meine Frage ist nun ob es eine Moeglichkeit gibt in meinem Fall Objekte zu versenden !?

if( key.isWritable() )
{
   SocketChannel channel = ( SocketChannel ) key.channel();

   String s = new String( "Test" );

   OutputStream os = Channels.newOutputStream( channel );
   ObjectOutputStream oos = new ObjectOutputStream( os );
   oos.writeObject( s );
   oos.flush();
}

 

[FArt]

Du solltest erst mal ein Tutorial durcharbeiten... Rox Java NIO Tutorial

 

[Teots]

Das hab ich bereits... deswegen hab ich auch nur den relevanten Codeabschnitt gepostet.

In allen Tutorials die ich bisher gelesen habe, kamen immer nur Beispiele mit ByteBuffern vor. Aber mir ist nicht klar wie ich ein beliebiges Objekt in einenen ByteBuffer ueberfuehren kann und dann daraus auch wieder ein Objekt erzeugen.

///////////// Nachtrag //////////////
Habs nun geloest... es reichte einfach aus, das Objekt, welches versendet werden sollte, mauell zu serialisieren. Also hier die Loesung fuer alle, die vllt. mal vor dem selben Problem stehen:

String s = new String( "Test" );

ByteArrayOutputStream bStream = new ByteArrayOutputStream();
ObjectOutputStream oStream = new ObjectOutputStream( bStream );
oStream.writeObject( s );
byte[] byteArray = bStream.toByteArray();

ByteBuffer objOutput = ByteBuffer.allocate( 2048 );
objOutput.clear();
objOutput.put( byteArray );
objOutput.flip();

int numWrite = channel.write( objOutput );

Zum Lesen des Objektes:

/* Get the channel associated with the key. */
SocketChannel channel = ( SocketChannel ) key.channel();

/* Create the input buffer an prepare it for reading. */
ByteBuffer in = ByteBuffer.allocate( 4096 );
in.clear();

/* Read the input. Save how much was read. */
int numRead = channel.read( in );

ByteArrayInputStream bis = new ByteArrayInputStream( in.array() );
ObjectInput oi = new ObjectInputStream( bis );
try
{
   String s = ( String ) oi.readObject();
}
catch( ClassNotFoundException e )
{
   // TODO Auto-generated catch block
   e.printStackTrace();
}

Wenn noch wer einen Vorschlag hat, wie es besser/schoener/schneller/einfacher geht... immer her damit

 

[derDude]

Servus,

wie würdest du es angehen wenn du größer Objekte als 4096 Byts empfangen möchtest, ich hab mich grade an ein dein Beispiel gehalten und muss feststellen das es so leider für Größere Objekte nicht funktioniert.

Hast du da eine Idee?

 

[tuxedo]

Da bedarf es eines Protokolls, das die Informationen über die größe mitüberträgt.

 

[derDude]

Müsste doch aber eigentlich auch ohne die Größe gehen.Ich wollte wollte wirklich einmal "Keep it f**king simple" nutzen. Nach dem diesem Motto würde ich vorgehen wollen was für mich bedeutet ich suche eine Lösung die in "Pseudo Code" so aussieht.

ByteArrayOutputStream zwischenSpeicher
Lese solange Stream/Channel nicht leer {
    Buffer = aktuellen Stream/Channel inhalt
    zwischenSpeicher.write(Buffer)
}
Object o = new ObjectInputStream(new ByteArrayInputStream(zwischenSpeicher.toByteArray())).readObject();
.... mit dem Objekt arbeiten

Irgend wie so muss das doch auch gehen. Ich hab nur noch keine Ahnung wie genau ....

 

[tuxedo]

Das Problem dabei ist ja: Wann weisst du dass du genug Daten empfangen hast?

Ein Protokoll muss nicht riesig oder aufwendig sein.

Du kannst das Objekt ja prima in einen ByteArrayOutputStream serialisieren und weißt dann anhand der Array-Größe wie groß das Objekt ist.

Bevor du nun das Objekt über's Netzwerk jagst, schickst du einfach die größe des byte-Arrays als Integer-Wert in Form von 4 bytes, gefolgt vom eigentlichen Objekt.

Auf Empfängerseite liest du dann immer zuerst 4 bytes, formst daraus wieder einen Integer und weiß dann, wieviele bytes du noch empfangen musst, bevor du mit dem readObject() beginnen kannst. Und das beste daran ist: Du kannst deinen Puffer in exakt der größe allokieren wie du ihn brauchst.

- Alex

 

[Teots]

Danke @tuxedo !
Das ist echt mal nen guter Tip fuer die Praxis. Werd ich direkt mal einbauen, da ich bis jetzt einfach immer sehr grosse Buffer benutze.

 

[tuxedo]

Wenn du unterschiedliche Objekte sendest, kannst du natürlich noch weitere IDs, Strings, Parameter vorher "mitschicken" damit die Empfängerseite weiß, wie sie den Daten umgehen soll.

Aber dann kann ich natürlich auch gleich wieder zu sowas wie MINA oder NETTY und Co. raten. Denn die Kapsel ersten NIO ziemlich gut und sehr performant, und zweitens unterstützen sie solche Protokollarbeiten von Haus aus.

- Alex

 

[derDude]

Ich hab grade mal Netty probiert, extrem gierig!!!

Das Beispiel ObjectEcho Beispiel scheint dabei genau das zusein was ich(scheinbar wir ^^) brauche nur leider hab ich noch nicht raus gefunden wie ich von mir aus, also von aussen ein Objekt an den Server senden kann.

Hier mal die Klassen um die es Geht

EchoObjectClient

import java.net.InetSocketAddress;
import java.util.concurrent.Executors;

import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
import org.jboss.netty.handler.codec.serialization.ObjectDecoder;
import org.jboss.netty.handler.codec.serialization.ObjectEncoder;

public class ObjectEchoClient {

	public static void main(String[] args) throws Exception {
		// Print usage if no argument is specified.
		if (args.length < 2 || args.length > 3) {
			System.err.println("Usage: "
					+ ObjectEchoClient.class.getSimpleName()
					+ " <host> <port> [<first message size>]");
			return;
		}

		// Parse options.
		final String host = args[0];
		final int port = Integer.parseInt(args[1]);
		final int firstMessageSize;

		if (args.length == 3) {
			firstMessageSize = Integer.parseInt(args[2]);
		} else {
			firstMessageSize = 256;
		}

		// Configure the client.
		ClientBootstrap bootstrap = new ClientBootstrap(
				new NioClientSocketChannelFactory(Executors
						.newCachedThreadPool(), Executors.newCachedThreadPool()));
		final ObjectEchoClientHandler client = new ObjectEchoClientHandler(firstMessageSize);
		// Set up the pipeline factory.
		bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
			public ChannelPipeline getPipeline() throws Exception {
				return Channels.pipeline(new ObjectEncoder(),
						new ObjectDecoder(), client);
			}
		});

		// Start the connection attempt.
		bootstrap.connect(new InetSocketAddress(host, port));
	}

}

und der EchoObjectClientHandler

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.jboss.netty.channel.ChannelEvent;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelState;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;

public class ObjectEchoClientHandler extends SimpleChannelUpstreamHandler {
	private static final Logger logger = Logger
			.getLogger(ObjectEchoClientHandler.class.getName());

	private final List<Integer> firstMessage;
	private final AtomicLong transferredMessages = new AtomicLong();

	/**
	 * Creates a client-side handler.
	 */
	public ObjectEchoClientHandler(int firstMessageSize) {
		if (firstMessageSize <= 0) {
			throw new IllegalArgumentException("firstMessageSize: "
					+ firstMessageSize);
		}
		firstMessage = new ArrayList<Integer>(firstMessageSize);
		for (int i = 0; i < firstMessageSize; i++) {
			firstMessage.add(Integer.valueOf(i));
		}
	}

	public long getTransferredMessages() {
		return transferredMessages.get();
	}

	public void handleUpstream(ChannelHandlerContext ctx, ChannelEvent e)
			throws Exception {
		if (e instanceof ChannelStateEvent
				&& ((ChannelStateEvent) e).getState() != ChannelState.INTEREST_OPS) {
			logger.info(e.toString());
		}
		super.handleUpstream(ctx, e);
	}

	@Override
	public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) {
		// Send the first message if this handler is a client-side handler.
		e.getChannel().write(firstMessage);
	}

	@Override
	public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) {
		// Echo back the received object to the client.
		transferredMessages.incrementAndGet();
		this.logger.info("Test :"+transferredMessages.toString());
		e.getChannel().write(e.getMessage());
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) {
		logger.log(Level.WARNING, "Unexpected exception from downstream.", e
				.getCause());
		e.getChannel().close();
	}
}

Soweit ist ja klar das beim Initialisieren der Verbindung ein Objekt an den vom Typ AtomicLong an den Server gesendet wird dieser sendet es wieder zurück es wird incrementiert und wieder an den Server gesendet.

Wie kann ich also aus dem ObjectEchoClient ein beliebiges Objekt senden?

 

[tuxedo]

Kann dir in Sachen Netty leider nicht wirklich weiterhelfen. Hab mich bisher nur auf MINA konzentriert. Und da finde ich die Samples etwas "einfacher" als die von Netty. Aber das ist jetzt mehr subjektiv als objektiv.

- Alex

 

[derDude]

Hier mal ein Komplettes beispiel für Netty:

import java.net.InetSocketAddress;
import java.util.concurrent.Executors;

import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
import org.jboss.netty.handler.codec.serialization.ObjectDecoder;
import org.jboss.netty.handler.codec.serialization.ObjectEncoder;

public class Server {

	/**
	 * @param args
	 */
	public static void main(String[] args) throws Exception {
		// Configure the server.
		ServerBootstrap bootstrap = new ServerBootstrap(
				new NioServerSocketChannelFactory(
						Executors.newCachedThreadPool(),
						Executors.newCachedThreadPool()));

		// Set up the pipeline factory.
		bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
			public ChannelPipeline getPipeline() throws Exception {
				return Channels.pipeline(new ObjectEncoder(),
						new ObjectDecoder(), new ServerHandler());
			}
		});

		// Bind and start to accept incoming connections.
		bootstrap.bind(new InetSocketAddress(8080));
		System.out.println("Server läuft ..");
	}

}

import java.util.logging.Level;
import java.util.logging.Logger;

import org.jboss.netty.channel.ChannelEvent;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelState;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;

public class ServerHandler extends SimpleChannelUpstreamHandler {

	private static final Logger logger = Logger.getLogger(ServerHandler.class
			.getName());
	
	@Override
	public void handleUpstream(ChannelHandlerContext ctx, ChannelEvent e)
			throws Exception {
		if (e instanceof ChannelStateEvent
				&& ((ChannelStateEvent) e).getState() != ChannelState.INTEREST_OPS) {
			logger.info(e.toString());
		}
		super.handleUpstream(ctx, e);
	}

	@Override
	public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) {
	}
	
	@Override
	public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) {
		if(e.getMessage() instanceof String){
			System.out.println("Nachricht erhalten: "+e.getMessage().toString());
		}
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) {
		logger.log(Level.WARNING, "Unexpected exception from downstream.",
				e.getCause());
		e.getChannel().close();
	}

}

Das war die Server seite. Der macht nix anderes als das er im Fall das ihm ein String zugesandt wurde er diesen ausgibt.

Nun der Client man beachte die Zeile nach dem Initialisieren der Verbindung.

import java.net.InetSocketAddress;
import java.util.concurrent.Executors;

import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
import org.jboss.netty.handler.codec.serialization.ObjectDecoder;
import org.jboss.netty.handler.codec.serialization.ObjectEncoder;

public class Client {

	public static void main(String[] args) {
		String host = "localhost";
		int port = 8080;
		// Clienti initialisieren
		ClientBootstrap bootstrap = new ClientBootstrap(
				new NioClientSocketChannelFactory(
						Executors.newCachedThreadPool(),
						Executors.newCachedThreadPool()));
		ClientHandler client = new ClientHandler();
		bootstrap.setPipelineFactory(new ChannelPipelineFactory() {

			@Override
			public ChannelPipeline getPipeline() throws Exception {
				return Channels.pipeline(new ObjectEncoder(),
						new ObjectDecoder(), new ClientHandler());
			}

		});
		
		ChannelFuture future = bootstrap.connect(new InetSocketAddress(host, port));
		Channel channel = future.awaitUninterruptibly().getChannel();
		channel.write("kleiner Test");
	}

import java.util.logging.Level;
import java.util.logging.Logger;

import org.jboss.netty.channel.ChannelEvent;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelState;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;

public class ClientHandler extends SimpleChannelUpstreamHandler {

	private static final Logger logger = Logger.getLogger(ClientHandler.class
			.getName());
	
	@Override
	public void handleUpstream(ChannelHandlerContext ctx, ChannelEvent e)
			throws Exception {
		if (e instanceof ChannelStateEvent
				&& ((ChannelStateEvent) e).getState() != ChannelState.INTEREST_OPS) {
			logger.info(e.toString());
		}
		super.handleUpstream(ctx, e);
	}

	@Override
	public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) {
		
	}

	@Override
	public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) {
		
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) {
		logger.log(Level.WARNING, "Unexpected exception from downstream.",
				e.getCause());
		e.getChannel().close();
	}
}